一.背景介绍

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.

参考资料:http://docs.confluent.io/2.0.0/platform.html