Debezium SQL server 2016同步到kafka

写这篇文章的主要目的是因为在自己使用Debezium的时候遇见的问题，网上的案例又比较少，自己决定写一篇文章，希望可以帮助到其他人，因为是第一次写博客，有什么不足的地方希望见谅或提出宝贵意见。

一、需求和软件的版本

（1）需求：监控sellInfo表的部分列，当这些列新增或更新或删除的时候，kafka生成一条变更信息到消费端，业务代码拿到这些信息来更新Elasticsearch

（2）软件版本：sqlServer版本，Microsoft SQL Server 2016 (SP1) ；

kafka版本，kafka2.12-2.5.0；

Debezium,JAR版本，debezium-debezium-connector-sqlserver-1.2.1

二、开启CDC

（1）为数据库开启CDC（需要sysadmin权限）

EXEC sys.sp_cdc_enable_db

（2）为数据库表开启CDC

EXEC sys.sp_cdc_enable_table

@source_schema = 'dbo', --数据库名

@source_name = 'SellInfo', --数据库表名

@capture_instance = default, --默认实例

@role_name = NULL, --角色权限，null表示不设置权限

@captured_column_list = 'ArticleId,HtmlID,Title' --为哪些列开启监控

sellInfo开启cdc成功的标志，数据库出现cdc库，如图所示，这是发生update操作时，cdc会做记录。

二、安装kafka和开启kafka-connect

（1）下载kafka，并解压,重命名

当前路径 /root

tar zxvf kafka_2.12-2.5.0.tgz -C ./

mv kafka_2.12-2.5.0 kafka

(2)修改kafka/config/server.properties

listeners = PLAINTEXT://192.168.16.2:9092

advertised.listeners=PLAINTEXT://192.168.16.2:9092

（3)开启zookeeper，开启kafka

当前路径 /root/kafka

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

bin/kafka-server-start.sh -daemon config/server.properties

(4)下载Debezium JAR包

JAR下载地址，下载jar，并解压到 /kafka/kafka_connect_plugins(自己创建一个文件夹)

（5)修改kafka/config/connect-distribute.properties

bootstrap.servers=192.168.16.2:9092

plugin.path= /root/kafka/kafka_connect_plugins

(6)启动kafka-connect

当前路径 /root/kafka

bin/connect-distributed.sh -daemon config/connect-distributed.properties

开启成功的标志,postman可以使用访问端口（默认的端口号,可以在connect-distributed.properties 更改）

(7)编辑connect的属性

当前路径 /root/kafka

创建一个json文件，vi sellInfo.json

{

"name": "sqlserver-cdc-dbo-sellInfo",

"config": {

"connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",

"database.hostname": "xx.xx.xx.xx", ---数据库ip

"database.port": "1433", ----端口号

"database.user": "xxxx", ----数据库用户名

"database.password": "xxxx", ----数据库密码

"database.dbname": "idea_test", -----数据库名

"database.server.name": "fullfillment",

"table.whitelist": "dbo.SellInfo", -----数据库表白名单

"snapshot.mode":"schema_only", ----快照方式,inital--表数据全量更新，schema_only--表数据开启cdc后的更新

"column.blacklist":"dbo.SellInfo.xxx,dbo.SellInfo.xxx" ----不需要监控的表的字段（下面有我用的时候遇见的问题）

"database.history.kafka.bootstrap.servers": "192.168.16.2:9092",

"database.history.kafka.topic": "dbhistory.fullfillment",

"value.converter.schemas.enable":"false",

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

"event.processing.failure.handing.mode":"warn"

}

更多数据设置可以参考Debezium的官方文档：

用于SQL Server的Debezium连接器（里面还有一些其他的连接器，mysql等）

(8)执行连接器

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://192.168.16.2:8083/connectors/ -d @sellInfo.json

开启成功的标志：

更多postman关于连接器的操作可以参考这篇文章，帮助我很多

sqlserver增量订阅&消费实时同步kafka，最新解决方案，看完不会你打我！

三、验证成果

1、update [dbo].SellInfo set Audit = 2 where ArticleId = 1 （可以是新增，可以是删除）

2、查看当前的有哪些topic

此时路径 /root/kafka

bin/kafka-topics.sh --zookeeper localhost --list

__consumer_offsets ----默认创建

connect-configs ----默认创建

connect-offsets ----默认创建

connect-status ----默认创建

dbhistory.fullfillment ----记录表的结构信息

fullfillment ----记录表的结构信息

fullfillment.dbo.SellInfo -----SellInfo表改变的记录

查看fullfillment.dbo.SellInfo的信息:

当前路径 /root/kafka

bin/kafka-console-consumer.sh --bootstrap-server 192.168.16.2:9092 --topic fullfillment.dbo.SellInfo --from-beginning

四、总结

（1）第一次搞这种冷门的东西，刚开始确实有点头疼，后来明白，一定要学会参考别人的东西，查看官方文档，制定流程分阶段进行等，这也是我为什么要写这篇文章的原因

（2）可能出现的问题，CDC的开启，如果在执行命令的时候出现错误、按照给出的错误信息去baidu；kafka的安装与运行 --daemon是可以让程序离开shell界面后台运行，但是不会显示开启时的日志；kafka-connect的配置文件sellInfo.json中的'column.blacklist'字段出现问题，比如一张表a，id,name,age,phone,我CDC监控id,age ,’column.blacklist‘剩下的字段，运行connect同步数据时会出错，我看了源码感觉应该是一个bug，当然很有可能是有些配置不知道怎么使用导致的，最后我通过分析源码将表的结构变成id,age,name,phone避免了这种异常，错误代码我贴出来，希望有人可以解答这个问题。

08-03 12:41:56,160] ERROR Error requesting a row value, row: 3, requested index: 15 at position 3 (io.debezium.relational.TableSchemaBuilder:221) [2020-08-03 12:41:56,160] ERROR Producer failure (io.debezium.pipeline.ErrorHandler:31) org.apache.kafka.connect.errors.ConnectException: Error while processing event at offset {transaction_id=null, event_serial_no=2, commit_lsn=000003cb:000ec557:0003, change_lsn=000003cb:000ec557:0002} at io.debezium.pipeline.EventDispatcher.dispatchDataChangeEvent(EventDispatcher.java:220) at io.debezium.connector.sqlserver.SqlServerStreamingChangeEventSource.lambda$execute$1(SqlServerStreamingChangeEventSource.java:247) at io.debezium.jdbc.JdbcConnection.prepareQuery(JdbcConnection.java:524) at io.debezium.connector.sqlserver.SqlServerConnection.getChangesForTables(SqlServerConnection.java:190) at io.debezium.connector.sqlserver.SqlServerStreamingChangeEventSource.execute(SqlServerStreamingChangeEventSource.java:162) at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:108) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) Caused by: org.apache.kafka.connect.errors.ConnectException: Data row is smaller than a column index, internal schema representation is probably out of sync with real database schema at io.debezium.relational.TableSchemaBuilder.validateIncomingRowToInternalMetadata(TableSchemaBuilder.java:222) at io.debezium.relational.TableSchemaBuilder.lambda$createValueGenerator$5(TableSchemaBuilder.java:251) at io.debezium.relational.TableSchema.valueFromColumnData(TableSchema.java:143) at io.debezium.relational.RelationalChangeRecordEmitter.emitUpdateRecord(RelationalChangeRecordEmitter.java:97) at io.debezium.relational.RelationalChangeRecordEmitter.emitChangeRecords(RelationalChangeRecordEmitter.java:51) at io.debezium.pipeline.EventDispatcher.dispatchDataChangeEvent(EventDispatcher.java:193) ... 10 more