摘要: 作者：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 --password="mypassword" 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;
注意上面第一行，--password="mypassword" ,对，你没看错，这里虚拟机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.