一. Kafka Connect简介

　　Kafka是一个使用越来越广的消息系统，尤其是在大数据开发中（实时数据处理和分析）。为何集成其他系统和解耦应用，经常使用Producer来发送消息到Broker，并使用Consumer来消费Broker中的消息。Kafka Connect是到0.9版本才提供的并极大的简化了其他系统与Kafka的集成。Kafka Connect运用用户快速定义并实现各种Connector(File,Jdbc,Hdfs等)，这些功能让大批量数据导入/导出Kafka很方便。

             

如图中所示，左侧的Sources负责从其他异构系统中读取数据并导入到Kafka中；右侧的Sinks是把Kafka中的数据写入到其他的系统中。

二. 各种Kafka Connector

　　Kafka Connector很多，包括开源和商业版本的。如下列表中是常用的开源Connector

Connectors	References
Jdbc	Source, Sink
Elastic Search	Sink1, Sink2, Sink3
Cassandra	Source1, Source 2, Sink1, Sink2
MongoDB	Source
HBase	Sink
Syslog	Source
MQTT (Source)	Source
Twitter (Source)	Source, Sink
S3	Sink1, Sink2

　　商业版的可以通过Confluent.io获得

三. 示例

3.1 FileConnector Demo

　本例演示如何使用Kafka Connect把Source(test.txt)转为流数据再写入到Destination(test.sink.txt)中。如下图所示：

          

      本例使用到了两个Connector:

• FileStreamSource：从test.txt中读取并发布到Broker中
• FileStreamSink：从Broker中读取数据并写入到test.sink.txt文件中

　　其中的Source使用到的配置文件是${KAFKA_HOME}/config/connect-file-source.properties

name=local-file-source
connector.class=FileStreamSource
tasks.max=1
file=test.txt
topic=connect-test

　　其中的Sink使用到的配置文件是${KAFKA_HOME}/config/connect-file-sink.properties

name=local-file-sink
connector.class=FileStreamSink
tasks.max=1
file=test.sink.txt
topics=connect-test

　　Broker使用到的配置文件是${KAFKA_HOME}/config/connect-standalone.properties

bootstrap.servers=localhost:9092key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=trueinternal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000

 

3.2 运行Demo

　　需要熟悉Kafka的一些命令行，参考本系列之前的文章Apache Kafka系列(二) 命令行工具（CLI）

 3.2.1 启动Kafka Broker

[root@localhost bin]# cd /opt/kafka_2.11-0.11.0.0/
[root@localhost kafka_2.11-0.11.0.0]# ls
bin  config  libs  LICENSE  logs  NOTICE  site-docs
[root@localhost kafka_2.11-0.11.0.0]# ./bin/zookeeper-server-start.sh ./config/zookeeper.properties &
[root@localhost kafka_2.11-0.11.0.0]# ./bin/kafka-server-start.sh ./config/server.properties &

3.2.2 启动Source Connector和Sink Connector

[root@localhost kafka_2.11-0.11.0.0]# ./bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties 

3.3.3 打开console-consumer

./kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic connect-test

3.3.4 写入到test.txt文件中，并观察3.3.3中的变化

[root@Server4 kafka_2.12-0.11.0.0]# echo 'firest line' >> test.txt
[root@Server4 kafka_2.12-0.11.0.0]# echo 'second line' >> test.txt
3.3.3中打开的窗口输出如下
{"schema":{"type":"string","optional":false},"payload":"firest line"}
{"schema":{"type":"string","optional":false},"payload":"second line"}

3.3.5 查看test.sink.txt

[root@Server4 kafka_2.12-0.11.0.0]# cat test.sink.txt 
firest line
second line

 本例仅仅演示了Kafka自带的File Connector，后续文章会完成JndiConnector，HdfsConnector，并且会使用CDC（Changed Data Capture）集成Kafka来完成一个ETL的例子

 

四. kafka 0.9 connect JDBC测试

kafka 0.9的connect功能，测试过程如下：

1.创建容器（本次采用docker容器构建kafka环境）

docker run -p 10924:9092 -p 21814:2181 --name confluent -i -t -d java /bin/bash

2.将confluent安装程序拷贝进容器；

docker cp  confluent.zip confluent:/root

3.进入到confluent容器

docker exec -it confluent /bin/bash

4.解压confluent压缩包

unzip confluent.zip

5.启动kafka

/root/confluent/bin/zookeeper-server-start  /root/confluent/etc/kafka/zookeeper.properties  & > zookeeper.log

/root/confluent/bin/kafka-server-start  /root/confluent/etc/kafka/server.properties & > server.log

/root/confluent/bin/schema-registry-start  /root/confluent/etc/schema-registry/schema-registry.properties & > schema.log

6.测试kafka 是否正常

开两个docker窗口，一个跑producer,一个跑consumer,

/root/confluent/bin/kafka-avro-console-producer --broker-list localhost:9092 --topic test --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}'

 

/root/confluent/bin/kafka-avro-console-consumer --topic test  --zookeeper localhost:2181 --from-beginning

在producer端依次输入以下记录，确认consumer能正确显示；

{"f1": "value1"}

{"f1": "value2"}

{"f1": "value3"}

以上为安装kafka过程，接下来开始测试jdbc接口；

测试之前，需要获取mysql JDBC的驱动并将获放在kafka环境对应的jre/lib文件夹里

测试jdbc connect

1.创建配置文件quickstart-mysql.properties，内容如下： 

name=test-mysql-jdbc-autoincrement
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:mysql://192.168.99.100:33061/test1?user=root&password=welcome1
mode=incrementing
incrementing.column.name=id
topic.prefix=test-mysql-jdbc-

注：mysql是我在另一个容器里运行的，jdbc:mysql://192.168.99.100:33061/test1?user=root&password=welcome1是连接容器里的mysql的连接串

2.执行./bin/connect-standalone etc/schema-registry/connect-avro-standalone.properties etc/kafka-connect-jdbc/quickstart-mysql.properties

3.执行./bin/kafka-avro-console-consumer --new-consumer --bootstrap-server 192.168.99.100:10924 --topic test-mysql-jdbc-accounts --from-beginning

然后在数据库里增加一条记录

然后就会在consumer端显示新增记录

 

五. 配置连接器

 

Connector的配置是简单的key-value映射。对于独立模式，这些都是在属性文件中定义，并通过在命令行上的Connect处理。在分布式模式，JSON负载connector的创建（或修改）请求。大多数配置都是依赖的connector，有几个常见的选项：

• name - 连接器唯一的名称，不能重复。
• connector.calss - 连接器的Java类。
• tasks.max - 连接器创建任务的最大数。
• connector.class配置支持多种格式：全名或连接器类的别名。比如连接器是org.apache.kafka.connect.file.FileStreamSinkConnector，你可以指定全名，也可以使用FileStreamSink或FileStreamSinkConnector。Sink connector也有一个额外的选项来控制它们的输入：
• topics - 作为连接器的输入的topic列表。

对于其他的选项，你可以查看连接器的文档。

六、rest api

kafka connect的目的是作为一个服务运行，默认情况下，此服务运行于端口8083。它支持rest管理，用来获取 Kafka Connect 状态，管理 Kafka Connect 配置，Kafka Connect 集群内部通信，常用命令如下：

GET /connectors 返回一个活动的connect列表
POST /connectors 创建一个新的connect；请求体是一个JSON对象包含一个名称字段和连接器配置参数

GET /connectors/{name} 获取有关特定连接器的信息
GET /connectors/{name}/config 获得特定连接器的配置参数
PUT /connectors/{name}/config 更新特定连接器的配置参数
GET /connectors/{name}/tasks 获得正在运行的一个连接器的任务的列表

DELETE /connectors/{name} 删除一个连接器，停止所有任务，并删除它的配置

GET /connectors 返回一个活动的connect列表

POST /connectors 创建一个新的connect；请求体是一个JSON对象包含一个名称字段和连接器配置参数

GET /connectors/{name} 获取有关特定连接器的信息
GET /connectors/{name}/config 获得特定连接器的配置参数
PUT /connectors/{name}/config 更新特定连接器的配置参数
GET /connectors/{name}/tasks 获得正在运行的一个连接器的任务的列表

DELETE /connectors/{name} 删除一个连接器，停止所有任务，并删除它的配置

curl -s <Kafka Connect Worker URL>:8083/ | jq   获取 Connect Worker 信息

curl -s <Kafka Connect Worker URL>:8083/connector-plugins | jq 列出 Connect Worker 上所有 Connector

curl -s <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/tasks | jq 获取 Connector 上 Task 以及相关配置的信息

curl -s <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/status | jq 获取 Connector 状态信息

curl -s <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/config | jq 获取 Connector 配置信息

curl -s -X PUT <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/pause 暂停 Connector

curl -s -X PUT <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/resume 重启 Connector

curl -s -X DELETE <Kafka Connect Worker URL>:8083/connectors/<Connector名字> 删除 Connector

创建新 Connector （以FileStreamSourceConnector举例）

1. curl -s -X POST -H "Content-Type: application/json" --data

2. '{"name": "<Connector名字>",

3. "config":

4. {"connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",

5. "key.converter.schemas.enable":"true",

6. "file":"demo-file.txt",

7. "tasks.max":"1",

8. "value.converter.schemas.enable":"true",

9. "name":"file-stream-demo-distributed",

10. "topic":"demo-distributed",

11. "value.converter":"org.apache.kafka.connect.json.JsonConverter",

12. "key.converter":"org.apache.kafka.connect.json.JsonConverter"}

13. }'

14. http://<Kafka Connect Worker URL>:8083/connectors | jq

 

更新 Connector配置 （以FileStreamSourceConnector举例）

1. curl -s -X PUT -H "Content-Type: application/json" --data

2. '{"connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",

3. "key.converter.schemas.enable":"true",

4. "file":"demo-file.txt",

5. "tasks.max":"2",

6. "value.converter.schemas.enable":"true",

7. "name":"file-stream-demo-distributed",

8. "topic":"demo-2-distributed",

9. "value.converter":"org.apache.kafka.connect.json.JsonConverter",

10. "key.converter":"org.apache.kafka.connect.json.JsonConverter"}'

11. <Kafka Connect Worker URL>:8083/connectors/file-stream-demo-distributed/config | jq

七、kafka connect + debezium，解析binlog至kafka

在已知kafka connect和debezium作用，会使用kafka的基础上，学会使用debezium来读取binlog，并通过kafka connect将读取的内容放入kafka topic中。 

基于kafka0.10.0和Debezium0.6,mysql5.6

kafka connect

• Kafka Connect是一种用于Kafka和其他数据系统之间进行数据传输的工具。
• 仅关注数据的复制，并且不处理其他任务
• Kafka connect有两个概念，一个source，另一个是sink。source是把数据从一个系统拷贝到kafka里，sink是从kafka拷贝到另一个系统里。
• 可使用插件，获取不同系统的数据。例如通过Debezium插件解析mysql的日志，获取数据。
• 支持集群，可以通过REST API管理Kafka Connect。
• 对数据的传输进行管理和监控。

Debezium

• Debezium是一个分布式平台，可将现有数据库转换为事件流，因此应用程序可以立即查看并立即响应数据库中每一行的更改。
• Debezium建立在Apache Kafka之上，并提供用于监视特定数据库管理系统的Kafka Connect兼容连接器。
• 本教程使用Debezium监控binlog。

准备操作

• mysql需开启binlog

1. [mysqld]

2. log-bin=mysql-bin #添加这一行就ok

3. binlog-format=ROW #选择row模式

4. server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复

• mysql需创建一个有mysql slave相关权限的账号，若mysql不在本机，则需要远程权限，防火墙放行。

1. //mysql slave相关权限

2. CREATE USER canal IDENTIFIED BY 'debezium';

3. GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%';

4. -- GRANT ALL PRIVILEGES ON *.* TO 'debezium'@'%' ;

5. FLUSH PRIVILEGES;

• 操作概述

• 安装并启动kafka
• 安装并启动mysql
• 下载Debezium的mysql连接器http://debezium.io/docs/install/并解压
• 安装debezium，即将解压目录写入classpath变量，例如：export classpath=/root/debezium-connector-mysql/* 
  只在当前shell有效
• 参考http://debezium.io/docs/connectors/mysql/的配置文件示例，写好配置文件。
• 以独立模式启动kafka connect，此时debezium会对数据库中的每一个表创建一个topic，消费相应的topic，即可获取binlog解析信息。

1. //启动kafka connect

2. bin/connect-standalone.sh config/connect-standalone.properties mysql.properties

3. //查看topic列表

4. bin/kafka-topics.sh --list --zookeeper localhost:2181

5. //消费该主题

6. bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

• 配置文件

1. //mysql.properties

2. name=inventory-connector

3. connector.class=io.debezium.connector.mysql.MySqlConnector

4. database.hostname=192.168.99.100

5. database.port=3306

6. database.user=debezium

7. database.password=dbz

8. database.server.id=184054

9. database.server.name=fullfillment

10. database.whitelist=inventory

11. database.history.kafka.bootstrap.servers=192.168.30.30:9092

12. database.history.kafka.topic=dbhistory.fullfillment

13. include.schema.changes=true

• 索引

debezium官网 http://debezium.io/ 

kafka文档 http://kafka.apache.org/0100/documentation.html

 

八、Kafka Connect的优点

1.对开发者提供了统一的实现接口
2.开发，部署和管理都非常方便，统一 
3.使用分布式模式进行水平扩展，毫无压力
4.在分布式模式下可以通过Rest Api提交和管理Connectors
5.对offset自动管理，只需要很简单的配置，而不像Consumer中需要开发者处理
6.流式/批式处理的支持

九、第三方资源

这是已经得到支持的组件，不需要做额外的开发： https://www.confluent.io/product/connectors/
括号中的Source表示将数据从其他系统导入Kafka，Sink表示将数据从Kafka导出到其他系统。
其他的我没看，但是JDBC的实现比较的坑爹，是通过primary key（如id）和时间戳（如updateTime）字段，

来判断数据是否更新，这样的话应用范围非常受局限。

 

十、Connector Development Guide

 

在kafka与其他系统间复制数据需要创建kafka connect,他们将数复制到kafka或者从kafka复制到其他系统

连接器有两种形式：sourceconnectors将另一个系统数据导入kafka，sinkconnectors将数据导出到另一个系统

连接器不执行任何数据复制：它们的描述复制的数据，并且负责将工作分配给多个task

task分为sourcetask与sinktask

每个task从kafka复制数据,connect会保证record与schema的一致性完成任务分配，通常record与schema的映射是明显的，每一个文件对应一个流，流中的每一条记录利用schema解析并且保存对应的offset,另外一种情况是我们需要自己完成这种映射，比如数据库，表的offset不是很明确（没有自增id),一种可能的选择是利用时间（timestamp)来完成增量查询。

Streams and Records

每一个stream是包含key value对的记录的序列，key value可以是原始类型，可以支持复杂结构，除了array,object，嵌套等。数据转换是框架来完成的，record中包含stream id与offset,用于定时offset提交，帮助当处理失败时恢复避免重复处理。

Dynamic Connectors

所有的job不是静态的，它需要监听外部系统的变化，比如数据库表的增加删除，当一个新table创建时，它必须发现并且更新配置由框架来分配给该表一个task去处理，当通知发布后框架会更新对应的task.

Developing a Simple Connector

例子很简单
在standalone模式下实现 SourceConnector/SourceTask 读取文件并且发布record给SinkConnector/SinkTask 由sink写入文件

Connector Example

我们将实现SourceConnector，SinkConnector实现与它非常类似，它包括两个私有字段存放配置信息（读取的文件名与topic名称）
public class FileStreamSourceConnector extends SourceConnector {
    private String filename;
    private String topic;
getTaskClass()方法定义实现执行处理的task
@Override
public Class getTaskClass() {
    return FileStreamSourceTask.class;
}
下面定义FileStreamSourceTask，它包括两个生命周期方法start,stop
@Override
public void start(Map<String, String> props) {
    // The complete version includes error handling as well.
    filename = props.get(FILE_CONFIG);
    topic = props.get(TOPIC_CONFIG);
}
@Override
public void stop() {
    // Nothing to do since no background monitoring is required.
}
最后是真正核心的方法getTaskConfigs（）在这里我们仅处理一个文件，所以我们虽然定义了max task（在配置文件里）但是只会返回一个包含一条entry的list
@Override
public List<Map<String, String>> getTaskConfigs(int maxTasks) {
    ArrayList>Map<String, String>> configs = new ArrayList<>();
    // Only one input stream makes sense.
    Map<String, String> config = new Map<>();
    if (filename != null)
        config.put(FILE_CONFIG, filename);
    config.put(TOPIC_CONFIG, topic);
    configs.add(config);
    return configs;
}
即使有多个任务，这种方法的执行通常很简单。它只是要确定输入任务的数量，这可能需要拉取数据从远程服务，然后分摊。请注意，这个简单的例子不包括动态输入。在下一节中看到讨论如何触发任务的配置更新。

Task Example - Source Task

实现task，我们使用伪代码描述核心代码
public class FileStreamSourceTask extends SourceTask<Object, Object> {
    String filename;
    InputStream stream;
    String topic;
    public void start(Map<String, String> props) {
        filename = props.get(FileStreamSourceConnector.FILE_CONFIG);
        stream = openOrThrowError(filename);
        topic = props.get(FileStreamSourceConnector.TOPIC_CONFIG);
    }
    @Override
    public synchronized void stop() {
        stream.close()
    }
start方法读取之前的offset,并且处理新的数据，stop方法停止stream,下面实现poll方法
@Override
public List<SourceRecord> poll() throws InterruptedException {
    try {
        ArrayList<SourceRecord> records = new ArrayList<>();
        while (streamValid(stream) && records.isEmpty()) {
            LineAndOffset line = readToNextLine(stream);
            if (line != null) {
                Map sourcePartition = Collections.singletonMap("filename", filename);
                Map sourceOffset = Collections.singletonMap("position", streamOffset);
                records.add(new SourceRecord(sourcePartition, sourceOffset, topic, Schema.STRING_SCHEMA, line));
            } else {
                Thread.sleep(1);
            }
        }
        return records;
    } catch (IOException e) {
        // Underlying stream was killed, probably as a result of calling stop. Allow to return
        // null, and driving thread will handle any shutdown if necessary.
    }
    return null;
}
该方法重复执行读取操作，跟踪file offset,并且利用上述信息创建SourceRecord，它需要四个字段：source partition,source offset,topic name,output value(包括value及value的schema)

Sink Tasks

之前描述了sourcetask实现，sinktask与它完全不同，因为前者是拉取数据，后者是推送数据
public abstract class SinkTask implements Task {
public void initialize(SinkTaskContext context) { ... }
public abstract void put(Collection<SinkRecord> records);
public abstract void flush(Map<TopicPartition, Long> offsets);
put方法是最重要的方法，接收sinkrecords，执行任何需要的转换，并将其存储在目标系统。此方法不需要确保数据已被完全写入目标系统，然后返回。事实上首先放入缓冲，因此，批量数据可以被一次发送，减少对下游存储的压力。sourcerecords中保存的信息与sourcesink中的相同。flush提交offset，它接受任务从故障中恢复，没有数据丢失。该方法将数据推送至目标系统，并且block直到写入已被确认。的offsets参数通常可以忽略不计，但在某些情况保存偏移信息到目标系统确保一次交货。例如，一个HDFS连接器可以确保flush()操作自动提交数据和偏移到HDFS中的位置。

Resuming from Previous Offsets

kafka connect是为了bulk 数据拷贝工作，它拷贝整个db而不是拷贝某个表，这样会使用connnect的input或者output随时改变，source connector需要监听source系统的改变，当改变时通知框架（通过ConnectorContext对象）
举例
if (inputsChanged())
    this.context.requestTaskReconfiguration();
当接收到通知框架会即时的更新配置，并且在更新前确保优雅完成当前任务
如果一个额外的线程来执行此监控，该线程必须存在于连接器中。该线程不会影响connector。然而，其他变化也会影响task，最常见的是输入流失败在输入系统中，例如如果一个表被从数据库中删除。这时连接器需要进行更改，任务将需要处理这种异常。sinkconnectors只能处理流的加入，可以分配新的数据到task（例如，一个新的数据库表）。框架会处理任何kafka输入的改变，例如当组输入topic的变化因为一个正则表达式的订阅。sinktasks应该期待新的输入流，可能需要在下游系统创造新的资源，如数据库中的一个新的表。在这些情况下，可能会出现输入流之间的冲突（同时创建新资源），其他时候，一般不需要特殊的代码处理一系列动态流  

Dynamic Input/Output Streams

FileStream连接器是很好的例子，因为他们很简单的，每一行是一个字符串。实际连接器都需要具有更复杂的数据格式。要创建更复杂的数据，你需要使用kafka connector数据接口:Schema,Struct
Schema schema = SchemaBuilder.struct().name(NAME)
                    .field("name", Schema.STRING_SCHEMA)
                    .field("age", Schema.INT_SCHEMA)
                    .field("admin", new SchemaBuilder.boolean().defaultValue(false).build())
                    .build();
Struct struct = new Struct(schema)
                           .put("name", "Barbara Liskov")
                           .put("age", 75)
                           .build();
如果上游数据与schema数据格式不一致应该在sinktask中抛出异常