开发者社区> light-rain> 正文

使用EMR-Kafka Connect进行数据迁移

简介: 流式处理中经常会遇到Kafka与其他系统进行数据同步或者Kafka集群间数据迁移的情景。使用EMR Kafka Connect可以方便快速的实现数据同步或者数据迁移。本文介绍使用EMR Kafka Connect的REST API接口在Kafka集群间进行数据迁移。
+关注继续查看

1.背景

流式处理中经常会遇到Kafka与其他系统进行数据同步或者Kafka集群间数据迁移的情景。使用EMR Kafka Connect可以方便快速的实现数据同步或者数据迁移。

Kafka Connect是一种可扩展的、可靠的,用于在Kafka和其他系统之间快速地进行流式数据传输的工具。例如可以使用Kafka Connect获取数据库的binglog数据,将数据库的数据迁入Kafka集群,以同步数据库的数据,或者对接下游的流式处理系统。同时,Kafka Connect提供的REST API接口可以方便的进行Kafka Connect的创建和管理。
Kafka Connect分为standalone和distributed两种运行模式。standalone模式下,所有的worker都在一个进程中运行;相比之下,distributed模式更具扩展性和容错性,是最常用的方式,也是生产环境推荐使用的模式。

本文介绍使用EMR Kafka Connect的REST API接口在Kafka集群间进行数据迁移,使用distributed模式。

2.环境准备

创建两个EMR集群,集群类型为Kafka。EMR Kafka Connect安装在task节点上,进行数据迁移的目的Kafka集群需要创建task节点。集群创建好后,task节点上EMR Kafka Connect服务会默认启动,端口号为8083。

注意要保证两个集群的网路互通,详细的创建流程见创建集群

3.数据迁移

3.1准备工作

EMR Kafka Connect的配置文件路径为/etc/ecm/kafka-conf/connect-distributed.properties,可以查看所有的配置。修改配置项请参考组件参数配置

在源Kafka集群创建需要同步的topic,例如

1

另外,Kafka Connect会将offsets, configs和任务状态保存在topic中,topic名对应配置文件中的offset.storage.topic、config.storage.topic 和status.storage.topic三个配置项。默认的,Kafka Connect会自动的使用默认的partition和replication factor创建这三个topic。

3.2创建Kafka Connect

在目的Kafka集群的task节点(例如emr-worker-3节点),使用curl命令通过json数据创建一个Kafka Connect。

curl -X POST -H "Content-Type: application/json" --data '{"name": "connect-test", "config": { "connector.class": "EMRReplicatorSourceConnector", "key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter", "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter", "src.kafka.bootstrap.servers": "${src-kafka-ip}:9092", "src.zookeeper.connect": "${src-kafka-curator-ip}:2181", "dest.zookeeper.connect": "${dest-kafka-curator-ip}:2181", "topic.whitelist": "${source-topic}", "topic.rename.format": "${dest-topic}", "src.kafka.max.poll.records": "300" } }' http://emr-worker-3:8083/connectors

json数据中,name字段代表创建的connect的名称,此处为connect-test;config字段需要根据实际情况进行配置,其中的变量说明如下表

字段 说明
topic.whitelist 源Kafka集群中需要同步的topic,多个topic用逗号隔开,例如connect
topic.rename.format 可选配置项,目的Kafka集群中同步后的topic,默认值为${topic.whitelist}.replica。例如源topic为connect,同步后的topic为connect.replica
src.kafka.bootstrap.servers 源Kafka集群broker地址
src.zookeeper.connect 源Kafka集群安装了zookeeper服务的节点内网IP
dest.zookeeper.connect 目的Kafka集群安装了zookeeper服务的节点内网IP

3.3查看Kafka Connect

查看所有的Kafka Connect

2

查看创建的connect-test的状态

3

查看task的信息

4

3.4数据同步

在源Kafka集群创建需要同步的数据。

5

3.5查看同步结果

在目的Kafka集群消费同步的数据。

6

可以看到,在源Kafka集群发送的100000条数据已经迁移到了目的Kafka集群。

4.小结

本文介绍并演示了使用EMR kafka Connect在Kafka集群间进行数据迁移的方法,关于Kafka Connect更详细的使用请参考Kafka官网资料REST API使用


版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
使用 Kafka + Spark Streaming + Cassandra 构建数据实时处理引擎
Apache Kafka 是一个可扩展,高性能,低延迟的平台,允许我们像消息系统一样读取和写入数据。我们可以很容易地在 Java 中使用 Kafka。 Spark Streaming 是 Apache Spark 的一部分,是一个可扩展、高吞吐、容错的实时流处理引擎。
2960 0
Apache Carbondata接入Kafka实时流数据
1.导入carbondata依赖的jar包 将apache-carbondata-1.5.3-bin-spark2.3.2-hadoop2.7.2.jar导入$SPARKHOME/jars;或将apache-carbondata-1.
1439 0
使用Java connector消费ABAP系统的函数
使用Java connector消费ABAP系统的函数
23 0
Kafka修炼日志(二):Connect简明使用教程
Connect是Kafka 0.9版本新增的功能,可以方便的从其它源导入数据到Kafka数据流(指定Topic中),也可以方便的从Kafka数据流(指定Topic中)导出数据到其它源。
57 0
使用EMR-Kafka Connect进行数据迁移
流式处理中经常会遇到Kafka与其他系统进行数据同步或者Kafka集群间数据迁移的情景。使用EMR Kafka Connect可以方便快速的实现数据同步或者数据迁移。本文介绍使用EMR Kafka Connect的REST API接口在Kafka集群间进行数据迁移。
13490 0
Kafka数据迁移MaxCompute最佳实践
本文向您详细介绍如何使用DataWorks数据同步功能,将Kafka集群上的数据迁移到阿里云MaxCompute大数据计算服务。
3635 0
iOS网络编程之三——NSURLConnection的简单使用
iOS网络编程之三——NSURLConnection的简单使用
50 0
SQL Server 2005 系统数据介绍:dm_exec_connections
原文:SQL Server 2005 系统数据介绍:dm_exec_connections转载:http://msdn.microsoft.com/zh-cn/library/ms181509(SQL.90).aspx   sys.dm_exec_connections、sys.dm_exec_sessions 和 sys.dm_exec_requests 动态管理视图映射到sys.sysprocesses 系统表。
1726 0
centos安装gitlab并进行数据迁移
centos安装gitlab并进行数据迁移
3340 0
+关注
7
文章
0
问答
来源圈子
更多
E-MapReduce是构建于阿里云ECS弹性虚拟机之上,利用开源大数据生态系统,包括Hadoop和Spark,为用户提供集群、作业、数据等管理的一站式大数据处理分析服务。加入钉钉群聊阿里云E-MapReduce交流2群,点击进入查看详情 https://qr.dingtalk.com/action/joingroup?code=v1,k1,cNBcqHn4TvG0iHpN3cSc1B86D1831SGMdvGu7PW+sm4=&_dt_no_comment=1&origin=11
+ 订阅
相关文档: E-MapReduce
文章排行榜
最热
最新
相关电子书
更多
OceanBase 入门到实战教程
立即下载
阿里云图数据库GDB,加速开启“图智”未来.ppt
立即下载
实时数仓Hologres技术实战一本通2.0版(下)
立即下载