数据传输工具 —— Kafka Connect
1、什么是 kafka connect？
Kafka Connect 是一种用于在 kafka 和其他系统之间可扩展、可靠的流式传输数据的工具。它使得能够快速定义将大量数据集合移入和移出 kafka 的连接器变得简单。
Kafka Connect 可以获取整个数据库或从应用程序服务器收集指标到 kafka 主题，使数据可用于低延迟的流处理。
导出作业可以将数据从 kafka topic 传输到二次存储和查询系统，或者传递到批处理系统以进行离线分析。
2、功能
kafka connector 通用框架，提供统一的集成 API
同时支持分布式模式和单机模式
自动化的 offset 管理，开发人员不必担心错误处理的影响
rest 接口，用来查看和管理 kafka connectors
3、概念
Connectors：通过管理任务来处理数据流的高级抽象
Tasks：数据写入 kafka 和从 kafka 读出的实现
Workers：运行 connectors 和 tasks 的进程
Converters：kafka connect 和其他存储系统直接发送和接收数据之间转换数据
Connector 决定了数据要从哪里复制过来以及数据应该写到哪里去，一个 connector 实例是一个需要负责在 kafka 和其他系统之间复制数据的逻辑作业，connector plugin 是 jar 文件，实现了 kafka 定义的一些接口来完成特定的任务。
Task 是 kafka connect 数据模型的主角，每一个 connector 都会协调一系列的 task 去执行任务，connector 可以把一项工作分割成许多的 task，然后再把 task 分发到各个 worker 中去执行（分布式模式下），task 不自己保存自己的状态信息，而是交给特定的 kafka 主题去保存（config.storage.topic 和 status.storage.topic）。在分布式模式下有一个概念叫做任务再平衡（Task Rebalancing），当一个 connector 第一次提交到集群时，所有的 worker 都会做一个 task rebalancing 从而保证每一个 worker 都运行了差不多数量的工作，而不是所有的工作压力都集中在某个 worker 进程中，而当每个进程挂了之后也会执行 task rebalance。
Connectors 和 Tasks 都是逻辑工作单位，必须安排在进程中执行，而在 kafka connect 中，这些进程就是 workers，分别有两种 worker：standalone、distributed。生产中 distributed worker 表现很棒，因为它提供了可扩展性以及自动容错的功能，可以用一个 group.id 来启动很多 worker 进程，在有效的 worker 进程中它们会自动地去协调执行 connector 和 task，如果新加或者挂了一个 worker，其他的 worker 会检测到然后再重新分配 connector 和 task。
Converter 会把 bytes 数据转换为 kafka connect 内部的格式，也可以把 kafka connect 内部存储格式的数据变成 bytes，converter 对 connector 来说是解耦的，所以其他的 connector 都可以重用。例如使用了 avro converter，那么 jdbc connector 可以写 avro 格式的数据到 kafka，同时 hfds connector 也可以从 kafka 中读出 avro 格式的数据。
4、实战
启动 confluent
cd /app/confluent/bin
./confluent local start
使用 standalone 模式启动
# 启动 kafka connect
$CONFLUENT_HOME/bin/connect-standalone \
$CONFLUENT_HOME/etc/kafka/connect-standalone.properties \
connector1.properties [connector2.properties]
在 $CONFLUENT_HOME/etc/kafka 下存在很多配置文件
其中 connect-standalone.properties 是启动 connect 服务组件自身的配置，内容如下：
# kafka 服务
bootstrap.servers=localhost:9092
# 转换器
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
# 是否启用转换器
key.converter.schemas.enable=true
value.converter.schemas.enable=true
# 偏移量存储文件名
offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000
# 插件路径
plugin.path=/usr/share/java,/app/confluent/share/confluent-hub-components
# 默认端口为8083，需要修改端口时启动以下配置
# rest.port=8084
（1）标准 connect
启动一个带 FileSource 的 Connect
connect-file-source.properties 是一个 source connect 的模板配置，启用该配置就能够从指定文件中复制数据到 kafka 中，其默认的配置如下：
# connect 的名字
name=local-file-source
# 将文件读取到数据流中
connector.class=FileStreamSource
# 工作线程是 1 个
tasks.max=1
# 读取的文件名为 test.txt
file=test.txt
# 复制到的主题为 connect-test
topic=connect-test
启动 connect
$CONFLUENT_HOME/bin/connect-standalone \
$CONFLUENT_HOME/etc/kafka/connect-standalone.properties \
$CONFLUENT_HOME/etc/kafka/connect-file-source.properties
结果报错 Java 内存不足
关闭虚拟机，加大内存，重启服务器和 confluent，再次启动 connect，报错 8083 端口已被绑定
修改 connect-standalone.properties 配置中的端口为 8084 再启动，新的报错：不存在 source 配置文件中的指定的文件，在启动路径下创建文件，日志恢复正常
echo -e "foo\nbar\n" > $CONFLUENT_HOME/test.txt
可以通过 kafka tools 看到新增了主题 connect-test，写入了3条数据
往文件中写入数据，会报告又成功提交一次偏移量
# 写数据
/app/confluent# echo -e "foo1\nbar1\n" >> test.txt
# 日志
INFO WorkerSourceTask{id=local-file-source-0} Finished commitOffsets successfully in 1 ms (org.apache.kafka.connect.runtime.WorkerSourceTask:515)
...
然后可以看到主题中多了3条数据
启动带 FileSource 和 FileSink 的 Connect
connect-file-sink.properties 是一个 source connect 的模板配置，启用该配置就能够从指定文件中复制数据到 kafka 中，其默认的配置如下：
# connect 的名字
name=local-file-sink
# 从数据流中读取数据到文件中
connector.class=FileStreamSink
# 工作线程是 1 个
tasks.max=1
# 写入的文件是 test.sink.txt
file=test.sink.txt
# 读取数据的主题是 connect-test
topics=connect-test
启动 connect
$CONFLUENT_HOME/bin/connect-standalone \
$CONFLUENT_HOME/etc/kafka/connect-standalone.properties \
$CONFLUENT_HOME/etc/kafka/connect-file-source.properties \
$CONFLUENT_HOME/etc/kafka/connect-file-sink.properties
可以看到自动创建了 test.sink.txt 文件
同时可以看到 consumer 中多了一个 connect-local-file-sink ，偏移量为6（即已将6条数据都 sink 到了文件中）
（2）REST API
使用 Rest API 必须启动分布式模式，通过 Rest API 可以管理集群中的 connect 服务，默认端口是 8083。
GET /connectors - 返回所有正在运行的connector名。
POST /connectors - 新建一个connector;请求体必须是json格式并且需要包含name字段和config字段，name是connectors的名字，config是json格式，必须包含connector的配置信息。
GET /connectors/{name} - 获取指定connector的信息。
GET /connectors/{name}/config - 获取指定connector的配置信息。
在分布式模式下，有两种方式来配置 connector，第一种是类似 standalone 模式一样，写好配置文件，然后在启动时指定
$CONFLUENT_HOME/bin/connect-distributed \
$CONFLUENT_HOME/etc/kafka/connect-distributed.properties \
$CONFLUENT_HOME/etc/kafka/connect-file-source.properties \
$CONFLUENT_HOME/etc/kafka/connect-file-sink.properties
另外一种方式更加灵活，就是直接通过 Rest API 来对 connector 配置进行增删查。
查看 connectors
添加 connectors
查看某个 connector
这里指定的文件是相对路径，所以要在 $CONFLUENT_HOME/bin 路径下创建一个 test-distributed.txt 文件
cd $CONFLUENT_HOME/bin
echo -e "foo\nbar\n" > test-distributed.txt
可以看到出现了 connect-distributed 主题
添加 sink
从服务器可以看到产生了 sink 文件
删除 connector
再次往 test-distributed.txt 文件中追加数据，可以看到 connect-distributed 主题中的数据增加了，source connector 依然在工作，但是 sink connector 已经停止了，所以 test-distributed.sink.txt 文件中数据不再从主题中复制。
【注意】
如果要在脚本中处理，发起HTTP请求，可以使用 curl 工具，将请求的配置在 json 文件中，如：
curl -d @$CONFLUENT_HOME/connect-file-sink.json \
-H "Content-Type: application/json" \
-X POST http://localhost:8083/connectors
创建带有 Convert 的 connector
{
"name": "local-file-source",
"config": {
"connector.class": "FileStreamSource",
"tasks.max": 1,
"file": "test-transformation.txt",
"topic": "connect-transformation",
"transforms": "MakeMap,InsertSource",
"transforms.MakeMap.type": "org.apache.kafka.connect.transforms.HoistField$Value",
"transforms.MakeMap.field": "line",
"transforms.InsertSource.type": "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.InsertSource.static.field": "data_source",
"transforms.InsertSource.static.value": "test-file-source"
}
}
添加 connector（由于跟上述实验 name 一致，所以需要先删除或者换个 name）
创建 test-transformation.txt 文件，可以看到自动创建了 connect-transformation 主题
添加 sink
{
"name": "local-file-sink",
"config": {
"connector.class": "FileStreamSink",
"tasks.max": 1,
"file": "test-transformation.sink.txt",
"topics": "connect-transformation"
}
}
可以看到 sink 自动生成了 test-transformation.sink.txt 文件，并且内容不是 source 过来的原始数据，而是经过 convertor 处理后的带格式的数据
（3）MySQL Source、ESSink
演示将数据从 MySQL 复制到 kafka 中，再通过 kafka 将数据下沉到 ElasticSearch。这里 MySQL 是数据源，所以需要支持 MySQL 的 source connector，ES 是目标数据系统，所以需要支持 ES 的 sink connectors，可以从 https://www.confluent.io/hub/ 下载。
MySQL
MySQL 下载插件搜索关键字 "JDBC"，可以看到提供了在线安装的脚本和离线安装的包下载。
MySQL 环境准备
# 安装 MySQL
sudo apt-get install mysql-server
# 安装 Confluent 插件
confluent-hub install confluentinc/kafka-connect-jdbc:10.4.1
# 将 MySQL 驱动上传到 confluent 目录
# mv mysql.jar /app/confluent/share/confluent-hub-components/confluentinc-kafka-connect-jdbc/lib
【注意】下载下来的 jdbc connector 插件，在处理 mysql 时需要相应的驱动，而插件不带驱动，实际采集数据时会报错，这时需要将驱动 jar 包拷贝到插件库目录中。
数据准备，创建用户并授权，用该用户创建数据库、表和插入数据
grant all on *.* to hyh@'localhost' identified by 'hyh';
create database studentsDB;
use studentsDB;
create table students (rollno int primary key auto_increment, name varchar(30), marks varchar(30));
insert into students (name, marks) values ('James', 35);
创建 source 配置文件（connect-mysql-source.properties），内容如下：
name=test-source-mysql-jdbc-autoincrement
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:mysql://127.0.0.1:3306/studentsDB?user=hyh&password=hyh
mode=incrementing
# 表中的自增列字段
incrementing.column.name=rollno
# 表会被采集到的 topic 名前缀，比如表名叫 students，对应的 topic 就为 test-mysql-jdbc-students
topic.prefix=test-mysql-jdbc-
启动 mysql source connector
$CONFLUENT_HOME/bin/connect-standalone \
$CONFLUENT_HOME/etc/kafka/connect-standalone.properties \
$CONFLUENT_HOME/etc/kafka/connect-mysql-source.properties
可以看到启动之后，开启了 JDBC source task，然后执行了查询的 SQL，最后提交和刷新的偏移量
与此同时，可以看到 kafka 中新增了一个 topic test-mysql-jdbc-students
里面有一条数据，如果此时往表中再插入两条数据，可以看到数据变成了3条
ElasticSearch
ES 下载插件搜索关键字 "ElasticSearch" ，可以看到有 ElasticSearch Sink Connector、ElasticSearch Source Connector，注意有些插件是支持 source、sink，有些是分开两个插件。
ES 环境准备
tar -zxvf elasticsearch-7.6.0-linux-x86_64.tar.gz -C /app
mv /app/elasticsearch-7.6.0 /app/elasticsearch
# 配置环境变量
export ES_HOME=/app/elasticsearch
export PATH=${ES_HOME}/bin:$PATH
# 安装 Confluent 插件
confluent-hub install confluentinc/kafka-connect-elasticsearch:13.0.0
启动 ES
cd /app/elasticsearch
.bin/elasticsearch
报错不能以root用户启动
创建用户用户组es，并修改 es 安装目录所属用户和组
chown -R es:es elasticsearch/
再次启动看到以下日志即正常
配置 sink 配置文件（connect-es-sink.properties），内容如下：
name=test-sink-elastic
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=1
connection.url=http://localhost:9200
topics=test-mysql-jdbc-students
key.ignore=true
type.name=kafka-connect
启动 ES sink connector
$CONFLUENT_HOME/bin/connect-standalone \
$CONFLUENT_HOME/etc/kafka/connect-standalone.properties \
$CONFLUENT_HOME/etc/kafka/connect-mysql-source.properties \
$CONFLUENT_HOME/etc/kafka/connect-es-sink.properties
访问 es 9092 端口查询数据，可以查到有三条数据
# 查询命令
curl -H "Content-Type: application/json" -X GET http://localhost:9200/test-mysql-jdbc-students/_search
# 查到的结果
{
"took": 121,
"timed_out": false,
"_shards": {
"total": 1,
"successful": 1,
"skipped": 0,
"failed": 0
},
"hits": {
"total": {
"value": 3,
"relation": "eq"
},
"max_score": 1,
"hits": [
{
"_index": "test-mysql-jdbc-students",
"_type": "_doc",
"_id": "test-mysql-jdbc-students+0+0",
"_score": 1,
"_source": {
"rollno": 1,
"name": "James",
"marks": "35"
}
},
{
"_index": "test-mysql-jdbc-students",
"_type": "_doc",
"_id": "test-mysql-jdbc-students+0+1",
"_score": 1,
"_source": {
"rollno": 2,
"name": "James2",
"marks": "36"
}
},
{
"_index": "test-mysql-jdbc-students",
"_type": "_doc",
"_id": "test-mysql-jdbc-students+0+2",
"_score": 1,
"_source": {
"rollno": 3,
"name": "James3",
"marks": "37"
}
}
]
}
}
往数据库插入一条新的数据
insert into students (name, marks) values ('James4', 38);
可以看到 es 侧接收到了这条数据
还没有任何评论，你来说两句吧