作者:
云魄,阿里云E-MapReduce 高级开发工程师,专注于流式计算,Spark Contributor
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,例如
另外,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。
json数据中,name字段代表创建的connect的名称,此处为connect-test;config字段需要根据实际情况进行配置,其中的变量说明如下表
3.3查看Kafka Connect
查看所有的Kafka Connect
查看创建的connect-test的状态
查看task的信息
3.4数据同步
在源Kafka集群创建需要同步的数据。
3.5查看同步结果
在目的Kafka集群消费同步的数据。
可以看到,在源Kafka集群发送的100000条数据已经迁移到了目的Kafka集群。
4.小结
本文介绍并演示了使用EMR kafka Connect在Kafka集群间进行数据迁移的方法,关于Kafka Connect更详细的使用请参考Kafka官网资料和REST API使用。