kafka connect_kafka connect下载博客

一. 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

　　商业版的可以通过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

6.测试kafka 是否正常

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

五. 配置连接器

六、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举例）

curl -s -X POST -H "Content-Type: application/json" --data 
 '{"name": "<Connector名字>", 
   "config":
    {"connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",
    "key.converter.schemas.enable":"true",
    "file":"demo-file.txt",
    "tasks.max":"1",
    "value.converter.schemas.enable":"true",
    "name":"file-stream-demo-distributed",
    "topic":"demo-distributed",
    "value.converter":"org.apache.kafka.connect.json.JsonConverter",
    "key.converter":"org.apache.kafka.connect.json.JsonConverter"}
 }' 
http://<Kafka Connect Worker URL>:8083/connectors | jq

更新 Connector配置 （以FileStreamSourceConnector举例）

curl -s -X PUT -H "Content-Type: application/json" --data 
'{"connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",
"key.converter.schemas.enable":"true",
"file":"demo-file.txt",
"tasks.max":"2",
"value.converter.schemas.enable":"true",
"name":"file-stream-demo-distributed",
"topic":"demo-2-distributed",
"value.converter":"org.apache.kafka.connect.json.JsonConverter",
"key.converter":"org.apache.kafka.connect.json.JsonConverter"}' 
<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

[mysqld]
log-bin=mysql-bin #添加这一行就ok
binlog-format=ROW #选择row模式
server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复

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

//mysql slave相关权限
CREATE USER canal IDENTIFIED BY 'debezium';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'debezium'@'%' ;
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解析信息。

//启动kafka connect
bin/connect-standalone.sh config/connect-standalone.properties mysql.properties
//查看topic列表
bin/kafka-topics.sh --list --zookeeper localhost:2181
//消费该主题
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

• 配置文件

    //mysql.properties
    name=inventory-connector
    connector.class=io.debezium.connector.mysql.MySqlConnector
    database.hostname=192.168.99.100
    database.port=3306
    database.user=debezium
    database.password=dbz
    database.server.id=184054
    database.server.name=fullfillment
    database.whitelist=inventory
    database.history.kafka.bootstrap.servers=192.168.30.30:9092
    database.history.kafka.topic=dbhistory.fullfillment
    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

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

Streams and Records

Dynamic Connectors

Developing a Simple Connector

Connector Example

Task Example - Source Task

Sink Tasks

Resuming from Previous Offsets

Dynamic Input/Output Streams

参考

Kafka Documentation

http://kafka.apache.org/documentation/#connect

https://www.confluent.io/product/connectors/