Kafka connect快速构建数据ETL通道

liuxiaoyun111

　　摘要: 作者：Syn良子 出处：http://www.cnblogs.com/cssdongl 转载请注明出处
　　业余时间调研了一下Kafka connect的配置和使用，记录一些自己的理解和心得,欢迎指正.
　　一.背景介绍
　　Kafka connect是Confluent公司(当时开发出Apache Kafka的核心团队成员出来创立的新公司)开发的confluent platform的核心功能.
　　大家都知道现在数据的ETL过程经常会选择kafka作为消息中间件应用在离线和实时的使用场景中,而kafka的数据上游和下游一直没有一个
　　无缝衔接的pipeline来实现统一,比如会选择flume或者logstash采集数据到kafka,然后kafka又通过其他方式pull或者push数据到目标存储.
　　而kafka connect旨在围绕kafka构建一个可伸缩的，可靠的数据流通道，通过kafka connect可以快速实现大量数据进出kafka从而和其
　　他源数据源或者目标数据源进行交互构造一个低延迟的数据pipeline.给个图更直观点，大家感受下.

　　二.Kafka-connect快速配置
　　这里Confluent官方很贴心的提供了一个集成的镜像以便quickstart,如下链接
　　https://s3-us-west-2.amazonaws.com/confluent-files/kafka_connect_blog.ova
　　这是存储在Amazon S3上的，直接点击即可下载.这里我使用VMWare直接打开,刚开始会提示一个错误，不用管它直接点击重试即可
　　系统加载的过程中会默认初始化虚拟机的网络配置，这里我建议提前设置好桥接网络，让该虚拟机使用桥接网络初始化.
　　加载成功后，登录进入该Ubuntu系统,默认的用户名和密码都是:vagrant.
　　然后ls查看vagrant用户目录，查看几个关键的脚本内容后,我分别介绍它们的功能
　　1>setup.sh:自动下载mysql,mysql jdbc driver,配置好mysql以及做为hive的metastore
　　2>start.sh:启动confluent platform,kafka,hadoop,hive相关服务
　　3>clean_up.sh:和start.sh相反的，会关闭掉所有的服务，而且还会删除掉所有的数据(例如hdfs namenode和 datanode的数据，其实相当于fs format了)
　　那么很明显，第一步肯定是执行setup.sh,这里执行后会报错如下

　　这里无法下载相关的软件包，好吧，那么我们需要更新一下下载源的索引，执行如下命令
　　sudo apt-get update
　　更新完毕后再次执行setup.sh安装好mysql,hive等服务
　　紧接着执行start.sh来启动上述服务，启动后应该有如下进程，这是一个伪分布式节点

　　对了，虚拟机各个服务(例如hive,zookeeper等),配置文件和日志文件在路径/mnt/下,组件的安装位置位于/opt下
　　三.Kafka connect快速使用
　　配置完以后就可以准备使用kafka-connect来快速构建一个数据pipeline了，如下图所示

　　整个过程是将数据以mysql作为数据源，将数据通过kafka connect快速ETL到hive中去.注意这里图中没画kafka
　　但是实际上是包含在kafka connect里面的,话不多说，开始使用
　　1>Mysql数据准备
　　执行如下命令

$ mysql -u root
mysql> CREATE DATABASE demo;
mysql> USE demo;
mysql> CREATE TABLE users (
->   id serial NOT NULL PRIMARY KEY,
->   name varchar(100),
->   email varchar(200),
->   department varchar(200),
->   modified timestamp default CURRENT_TIMESTAMP NOT NULL,
->   INDEX `modified_index` (`modified`)
-> );
mysql> INSERT INTO users (name, email, department) VALUES ('alice', 'alice@abc.com', 'engineering');
mysql> INSERT INTO users (name, email, department) VALUES ('bob', 'bob@abc.com', 'sales');
mysql> exit;　　注意上面第一行， ,对，你没看错，这里虚拟机mysql的root默认密码就是mypassword,
　　强迫症患者请自行更改.随后建库，建表，插入数据.
　　2>关键概念准备
　　这里我快速普及一下参考官方文档理解的一些关键概念.
　　kafka connector:kafka connector是kafka connect的关键组成部分，它是一个逻辑上的job,用于在kafka和其他系统之间拷贝数据,比如
　　从上游系统拷贝数据到kafka,或者从kafka拷贝数据到下游系统
　　Tasks:每个kafka connector可以初始化一组task进行数据的拷贝
　　Workers:逻辑上包含kafka connector和tasks用来调度执行具体任务的进程,具体执行时分为standalone模式和distributed模式
　　见下图，这个是kafka上游的数据stream过来后，定义好对应的kafka connector后，分解为一组tasks然后push数据到kafka的不同topic

　　3>利用Kafka-connect摄取数据
　　主要是通过配置来实现从mysql摄取数据到kafka,然后按照topic来获取数据写入hdfs,命令如下


connect-standalone /mnt/etc/connect-avro-standalone.properties \
/mnt/etc/mysql.properties /mnt/etc/hdfs.properties &　　注意上面这些properties文件是虚拟机已经事先配置好的，可以直接执行实现数据的摄取
　　当前使用的kafka connect的standalone模式，当然还有distributed模式后续可以尝试
　　上面的那条命令的格式是这样:
connect-standalone worker.properties connector1.properties [connector2.properties connector3.properties ...]
　　主要解释一下connect-standalone后面的参数
　　worker.properties:就是上面提到过的worker进程的配置文件,可以定义kafka cluster的相关信息以及数据序列化的格式.
　　随后的一些参数就是kafka connector的配置参数了,比如上面的mysql.properties定义了一个kafka jdbc connector,用来同步mysql数据到kafka
　　最后一个hdfs.properties是kafka hdfs connector的配置文件,用来消费kafka topic数据push到hdfs.
　　那么执行这条命令后就可以将mysql的数据通过kafka connect快速ETL到hdfs了.
　　最后可以通过hive创建外表映射hdfs上的数据文件,然后在hive中查看对应数据,如下


$ hive
hive> SHOW TABLES;
OK
test_jdbc_users
hive> SELECT * FROM test_jdbc_users;
OK
1 alice alice@abc.com engineering 1450305345000
2 bob   bob@abc.com   sales       1450305346000　　四.Kafka connect使用总结
　　1>Kafka connect的使用其实就是配置不同的kafka connectors,这里大家可以把kafka作为中间组件，然后可以类比flume理解,kafka上游的
　　connector其实就是fllume的source从上游数据源sink到kafka,kafka的下游connector其实就是flume的source是kafka,sink到下游系统.
　　2>Kafka connect的数据pipeline要打通，它要求数据遵守confluent自己的一套通用的schema机制，细心的同学会发现上面jps后会有个进程名
　　SchemaRegistryMain,这里官方默认使用Avro格式进出Kafka,所以要留意worker.properties文件的配置信息.
　　3>我在使用中没有发现Flume 相关的connector,因此很好奇它应该是没有实现上游flume conector的属性配置。问题应该出在Flume的数据是基
　　于event的,而和上面2中所说的schema定义格式没有很好的兼容.
　　4>kafka connect的distributed模式应该更实用，随后会尝试，以及confluent所支持的实时处理流kafka streams.
　　参考资料:http://docs.confluent.io/2.0.0/platform.html