2018-10-20 作者: 张子阳 分类: 大数据处理
在处理实时数据时，需要即时地获得数据库表中数据的变化，然后将数据变化发送到Kafka中。这篇文章将介绍如何使用Kafka Connector完成这一工作。当获取实时数据时，数据源需要支持对数据变化进行反馈。不同的数据源采用了不同的技术和方法实现该功能，因为我们的业务数据库是MS SQL Server，因此这篇文章采用MSQL作为数据源。

1. 选择Connector

首先需要选择Connector，不同的数据源有不同的Connector，例如ActiveMQ Connector、MySql Connector、MSSQL Connector等。即便是同一数据源，也可能有不同的第三方提供。我一共尝试了下面两个MSSQL Connector：

比较遗憾的是：这两个Connector，debezium的是Alpha版本，confluent的是Preview版本，反正都不是正式版，而MySql都已经有正式版本了，可见开源社区对MS真的不友好呀 >_<、 它们两个一个是使用MSSQL Server的 Change Data Capture 获取数据变更，一个是使用 Change Tracking。
因为Change Tracking相比Change Data Capture来说，更轻量一些，因此我选用了confluent的Connector。其下载地址是：https://www.confluent.io/hub/

2. 安装Connector

下载后，将其解压缩至 $KAFKA_HOME/connectors 文件夹下，如下图所示：

说明：$KAFKA_HOME是你的kafka安装目录，如果是集群，要安装在集群下每台机器的connectors目录下。 Kafka Connect 实时读取MSSQL数据到Kafka - 图1
图1.
在上面的截图中，可以看到我还安装了confluentinc-kafka-connect-hdfs-5.0.0和debezium-connector-sqlserver两个connector。

3. 配置Connector

接下来要对Connector进行配置，此时可以回顾一下 Kafka Connect 基本概念。Connector是一组独立的集群，并且是作为Kafka集群的客户端，我们首先需要对Connector进行配置，配置文件位于 $KAFKA_HOME/config/connect-distributed.properties：

  1. # kafka集群地址
  2. bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
  3. # Connector集群的名称，同一集群内的Connector需要保持此group.id一致
  4. group.id=connect-cluster
  5. # 存储到kafka的数据格式
  6. key.converter=org.apache.kafka.connect.json.JsonConverter
  7. value.converter=org.apache.kafka.connect.json.JsonConverter
  8. key.converter.schemas.enable=false
  9. value.converter.schemas.enable=false
  10. # 内部转换器的格式，针对offsets、config和status，一般不需要修改
  11. internal.key.converter=org.apache.kafka.connect.json.JsonConverter
  12. internal.value.converter=org.apache.kafka.connect.json.JsonConverter
  13. internal.key.converter.schemas.enable=false
  14. internal.value.converter.schemas.enable=false
  15. # 用于保存offsets的topic，应该有多个partitions，并且拥有副本(replication)
  16. # Kafka Connect会自动创建这个topic，但是你可以根据需要自行创建
  17. # 如果kafka单机运行，replication.factor设置为1；当kafka为集群时，可以设置不大于集群中主机数
  18. # 因为我这里的环境是3主机的集群，因此设为2
  19. offset.storage.topic=connect-offsets
  20. offset.storage.replication.factor=2
  21. offset.storage.partitions=12
  22. # 保存connector和task的配置，应该只有1个partition，并且有多个副本
  23. config.storage.topic=connect-configs
  24. config.storage.replication.factor=2
  25. # 用于保存状态，可以拥有多个partition和replication
  26. status.storage.topic=connect-status
  27. status.storage.replication.factor=2
  28. status.storage.partitions=6
  29. # Flush much faster than normal, which is useful for testing/debugging
  30. offset.flush.interval.ms=10000
  31. # RESET主机名，默认为本机
  32. #rest.host.name=
  33. # REST端口号
  34. rest.port=18083
  35. # The Hostname & Port that will be given out to other workers to connect to i.e. URLs that are routable from other servers.
  36. #rest.advertised.host.name=
  37. #rest.advertised.port=
  38. # 保存connectors的路径
  39. # plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
  40. plugin.path=/opt/kafka/kafka_2.11-1.1.0/connectors

注意到connect-distributed.properties中的distributed。Kafka Connector有两种运行模式，单机（Standalone）和分布式（Distrubited）。因为单机通常作为测试运行，因此这篇文章只演示分布式运行模式。在config文件夹下，还有一个单机运行的配置文件，叫做connect-standalone.properties，内容大同小异。

4. 创建Topic

尽管首次运行Kafka connector时，会自动创建上面的topic，但是如果创建出错，那么Connector就会启动失败。保险起见，可以在运行Connector之前，手动创建好上面的三个特殊topic。

  1. # bin/kafka-topics.sh --zookeeper zookeeper1:2181/kafka --create --topic connect-offsets --replication-factor 2 --partitions 12
  2. # bin/kafka-topics.sh --zookeeper zookeeper1:2181/kafka --create --topic connect-configs --replication-factor 2 --partitions 1
  3. # bin/kafka-topics.sh --zookeeper zookeeper1:2181/kafka --create --topic connect-status --replication-factor 2 --partitions 6

5. 运行Connector

接下来就可以运行Connctor了，此时还没有涉及到任何业务或者数据库相关的配置和操作（即 Kafka Connect 基本概念 中提到的用户配置）。
执行下面的代码以运行Connector：
# bin/connect-distributed.sh config/connect-distributed.properties
上面这样是前台运行，当退出shell后进程也就结束了，前台运行的好处就是在开始运行时便于调试。如果想要后台运行，则需加上-daemon选项：
# bin/connect-distributed.sh -daemon config/connect-distributed.properties
运行connect时，会看到不停地涌现大量INFO信息，此时可以修改一下connect-log4j.properties，只显示WARN信息。
# vim config/connect-log4j.properties
log4j.rootLogger=WARN, stdout

6. 开启MSSQL数据库的Change Tracking

在继续进行之前，我们在数据库中创建表test_online，并且开启Change Tracking功能：

  1. Go
  2. CREATE TABLE [dbo].[test_online](
  3.  [Id] [int] IDENTITY(1,1) NOT NULL,
  4.  [UserName] [varchar](50) NOT NULL,
  5.  [IsOnline] [bit] NOT NULL,
  6.  [LastLogin] [int] NOT NULL,
  7. CONSTRAINT [PK_test_online] PRIMARY KEY CLUSTERED 
  8. (
  9.  [Id] ASC
  10. )
  11. Go
  12. ALTER DATABASE db_name  
  13. SET CHANGE_TRACKING = ON  
  14. (CHANGE_RETENTION = 2 DAYS, AUTO_CLEANUP = ON)  
  15. Go
  16. ALTER TABLE [db_name].dbo.[test_online]  
  17. ENABLE CHANGE_TRACKING  
  18. WITH (TRACK_COLUMNS_UPDATED = ON)

7. Kafka Connector REST API

当Kafka Connector运行起来以后，它就开启了REST API端口，像我们上面配置的是：18083。如果我们需要运行Task，比如实时捕捉数据库数据变化并写入Kafka，那么就需要像这个REST API提交用户配置（User Config）。在提交用户配置之前，我们先看看Kafka Connector REST API都包含哪些常见功能：

7.1 获取Worker的信息

因为我的kafka（主机名分别为kafka1、kafka2、kafka3）和kafka connector集群是共用主机的，因此可以使用下面的命令获取（你需要将下面的kafka1改成ip或者相应的主机名）：
# curl -s kafka1:18083/ | jq
{
“version”: “1.1.0”,
“commit”: “fdcf75ea326b8e07”,
“kafka_cluster_id”: “N93UISCxTS-SYZPfM8p1sQ”
}

7.2 获取Worker上已经安装的Connector

此时的Connector是静态概念，即上面第一节安装的Confluent MSSQL Connector，从下面的显示可以看到，我安装了好几个Connector：

  1. # curl -s kafka1:18083/connector-plugins | jq
  2. [
  3.   {
  4.     "class": "io.confluent.connect.cdc.mssql.MsSqlSourceConnector",
  5.     "type": "source",
  6.     "version": "0.0.1.9"
  7.   },
  8.   {
  9.     "class": "io.confluent.connect.hdfs.HdfsSinkConnector",
  10.     "type": "sink",
  11.     "version": "5.0.0"
  12.   },
  13.   {
  14.     "class": "io.confluent.connect.hdfs.tools.SchemaSourceConnector",
  15.     "type": "source",
  16.     "version": "1.1.0"
  17.   },
  18.   {
  19.     "class": "io.confluent.connect.storage.tools.SchemaSourceConnector",
  20.     "type": "source",
  21.     "version": "1.1.0"
  22.   },
  23.   {
  24.     "class": "io.debezium.connector.sqlserver.SqlServerConnector",
  25.     "type": "source",
  26.     "version": "0.9.0.Alpha1"
  27.   },
  28.   {
  29.     "class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
  30.     "type": "sink",
  31.     "version": "1.1.0"
  32.   },
  33.   {
  34.     "class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  35.     "type": "source",
  36.     "version": "1.1.0"
  37.   }
  38. ]

对于你来说，可能就只有io.confluent.connect.cdc.mssql.MsSqlSourceConnector这一个connector。

7.3 列出当前运行的connector(task)

curl -s kafka1:18083/connectors | jq
[]
因为我们当前Connector中没有提交过任何的用户配置（即没有启动Task），因此上面返回空数组。

7.4 提交Connector用户配置

当提交用户配置时，就会启动一个Connector Task，Connector Task执行实际的作业。用户配置是一个Json文件，同样通过REST API提交：
# curl -s -X POST -H “Content-Type: application/json” —data ‘{
“name”: “connector-mssql-online”,
“config”: {
“connector.class”: “io.confluent.connect.cdc.mssql.MsSqlSourceConnector”,
“tasks.max”: 1,
“server.name”: “192.168.0.21”,
“server.port” : “1433”,
“username”: “user_id”,
“password”: “your_password”,
“initial.database”: “db_name”,
“change.tracking.tables”: “dbo.test_online”
}
}’ http://kafka1:18083/connectors | jq

注意上面的配置要修改成你的本地配置。提交完成后，再次执行上一小节的命令，会看到已经有一个connector在运行了，其名称为connector-mssql-online：
curl -s kafka1:18083/connectors | jq
[
“connector-mssql-online”
]

7.5 查看connector的信息

curl -s kafka1:18083/connectors/connector-mssql-online | jq
{
“name”: “connector-mssql-online”,
“config”: {
“connector.class”: “io.confluent.connect.cdc.mssql.MsSqlSourceConnector”,
“password”: “your_password”,
“initial.database”: “db_name”,
“server.name”: “192.168.0.21”,
“tasks.max”: “1”,
“server.port”: “1433”,
“name”: “connector-mssql-online”,
“change.tracking.tables”: “dbo.test_online”,
“username”: “user_id”
},
“tasks”: [
{
“connector”: “connector-mssql-online”,
“task”: 0
}
],
“type”: “source”
}

注意：上面task:0，不是说有0个task，是task的id是0。

7.6 查看connector下运行的task信息

使用下面的命令，可以查看connector下运行的task的信息：
# curl -s kafka1:18083/connectors/connector-mssql-online/tasks | jq
[
{
“id”: {
“connector”: “connector-mssql-online”,
“task”: 0
},
“config”: {
“connector.class”: “io.confluent.connect.cdc.mssql.MsSqlSourceConnector”,
“password”: “your_password”,
“initial.database”: “db_name”,
“task.class”: “io.confluent.connect.cdc.mssql.MsSqlSourceTask”,
“server.name”: “192.168.0.21”,
“tasks.max”: “1”,
“server.port”: “1433”,
“name”: “connector-mssql-online”,
“change.tracking.tables”: “dbo.test_online”,
“username”: “user_id”
}
}
]
这里task的配置信息继承自connector的配置。

7.7 查看connector当前状态

curl -s kafka1:18083/connectors/connector-mssql-online/status | jq
{
“name”: “connector-mssql-online”,
“connector”: {
“state”: “RUNNING”,
“worker_id”: “192.168.0.31:18083”
},
“tasks”: [
{
“state”: “RUNNING”,
“id”: 0,
“worker_id”: “192.168.0.31:18083”
}
],
“type”: “source”
}

7.8 暂停/重启 Connector

curl -s -X PUT kafka1:18083/connectors/connector-mssql-online/pause
# curl -s -X PUT kafka1:18083/connectors/connector-mssql-online/resume

7.9 删除 Connector

curl -s -X DELETE kafka1:18083/connectors/connector-mssql-online

8. 从Kafka中读取变动数据

默认情况下，MSSQL Connector会将表的变动写入到：${databaseName}.${tableName} 这个topic中，这个topic的名称可以通过 topic.format 这个用户配置参数中进行设置，因为我们并没有配置，因此，topic的名称为db_name.test_online。
运行下面的控制台脚本，从Kafka中实时读取topic的内容：
# bin/kafka-console-consumer.sh —bootstrap-server kafka1:9092 —topic db_name.test_online —from-beginning

此时因为没有任何数据，因此控制台会阻塞。

9. 对test_online表进行修改

依次执行下面的增删改语句，对test_online表进行修改：

  1. insert into [test_online](UserName,IsOnline,LastLogin)
  2. values('子阳', 1, DATEDIFF(s, '19700101',GETDATE()))
  3. update test_online Set UserName='吉米' where UserName='子阳'
  4. Delete test_online Where UserName='吉米'

现在查看Kafka读取端控制台，可以看到以Json格式实时收到了数据库变动的消息：

  1. # bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic tgstat_ddztest.test_online
  2. {"Id":5,"UserName":"子阳","IsOnline":true,"LastLogin":1540635666,"_cdc_metadata":{"sys_change_operation":"I","sys_change_creation_version":"13","sys_change_version":"13","databaseName":"tgstat_ddztest","schemaName":"dbo","tableName":"test_online"}}
  3. {"Id":5,"UserName":"吉米","IsOnline":true,"LastLogin":1540635666,"_cdc_metadata":{"sys_change_operation":"U","sys_change_creation_version":"0","sys_change_version":"14","databaseName":"tgstat_ddztest","schemaName":"dbo","tableName":"test_online"}}
  4. null

从上面的消息可以看到，对于delete操作，收到了null。对于insert和update操作，收到了详细的变动信息。
至此，我们就配置完了Kafka Connector，并且实时获取到了数据库变更的消息。后续可以使用Spark Stream连接至此Topic，进行实时的数据运算和分析。以后有机会再进行掩饰。
感谢阅读，希望这篇文章能给你带来帮助！