本篇文章给大家分享的是有关EMR-Kafka中怎么利用Connect实现数据迁移，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着小编一起来看看吧。
创新互联公司专注于东兰网站建设服务及定制，我们拥有丰富的企业做网站经验。 热诚为您提供东兰营销型网站建设，东兰网站制作、东兰网页设计、东兰网站官网定制、微信小程序开发服务，打造东兰网络公司原创品牌,更为您提供东兰网站排名全网营销落地服务。
流式处理中经常会遇到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模式。
创建两个EMR集群，集群类型为Kafka。EMR Kafka Connect安装在task节点上，进行数据迁移的目的Kafka集群需要创建task节点。集群创建好后，task节点上EMR Kafka Connect服务会默认启动，端口号为8083。
注意要保证两个集群的网路互通，详细的创建流程见创建集群https://help.aliyun.com/document_detail/28088.html
EMR Kafka Connect的配置文件路径为/etc/ecm/kafka-conf/connect-distributed.properties。
在源Kafka集群创建需要同步的topic，例如
另外，Kafka Connect会将offsets, configs和任务状态保存在topic中，topic名对应配置文件中的offset.storage.topic、config.storage.topic 和status.storage.topic三个配置项。默认的，Kafka Connect会自动的使用默认的partition和replication factor创建这三个topic。
在目的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 |
查看所有的Kafka Connect
查看创建的connect-test的状态
查看创建的connect-test的状态查看task的信息
3.4数据同步
在源Kafka集群创建需要同步的数据。
3.5查看同步结果
在目的Kafka集群消费同步的数据。
可以看到，在源Kafka集群发送的100000条数据已经迁移到了目的Kafka集群。
以上就是EMR-Kafka中怎么利用Connect实现数据迁移，小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注创新互联行业资讯频道。
