CDC系列（三）、Debezium 监控MySQL操作日志实时同步到Kafka（对比canal，maxwell）

我就是我 2022-12-10 10:20 881阅读 0赞

**目录**

前言

准备工作

配置Connector

注册启动Connector

验证CDC

尾巴

--------------------

**CDC系列：**

[CDC系列（一）、Canal 集群部署及使用（带WebUI）][CDC_Canal _WebUI]

[CDC系列（二）、Maxwell\_v1.27.1 监控MySQL操作日志实时同步到Kafka][CDC_Maxwell_v1.27.1 _MySQL_Kafka]

--------------------

# 前言 #

前两篇我们介绍了CDC工具，以及利用Canal和Maxwell监控mysql操作日志并写入kafka，本篇我们来讲解另一个CDC工具：debezium。debezium不仅仅支持mysql的监控，目前最新版已支持mysql,postgre,mongodb,oracle,sqlserver,db2,cassandra等，它依赖于kafka connect 服务，通过kafka connect，我们可以很方便的将debezium采集到的数据库日志输出到各类目标端。

本篇我们演示如何利用kafka connect启动debezium mysql connector监控mysql 数据库操作日志并写入kafka内

# 准备工作 #

上面提到debezium依赖于kafka connect ，而且kafka connect是在0.9版本之后才由confluent团队贡献出来，因此我们必须要先安装好kafka0.9+版本，如果没有的话也可以参考我之前的文章：

*  [kafka 2.4.1单机版部署及使用][kafka 2.4.1]
 *  [Kafka系列（一）、2.6.0版本kafka集群搭建][Kafka_2.6.0_kafka]
 *  [Kafka系列（七）、Kafka套件 Confluent Platform 单机/集群部署][Kafka_Kafka_ Confluent Platform]

由于我们要监控mysql的binlog，因此还需要一个可以访问mysql binlog权限的用户，这个操作在前面两篇有介绍就不在此赘述了。

# 配置Connector #

**1. 下载**，到官方网址下载最新1.2 版debezium-mysql-connector：[https://debezium.io/releases/1.2/][https_debezium.io_releases_1.2]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70][]

**2. 配置kafka connect服务**

1. 在kafka目录下新建一个文件夹
    mkdir /opt/app/lib_conn
    
    2. 将下载的debezium插件上传至服务器并解压至刚才创建的目录
    tar -zxvf debezium-connector-mysql-1.2.5.Final-plugin.tar.gz -C /opt/app/lib_conn
    
    3. 修改kafka connect 配置文件
    vim /opt/app/kafka/connect-standalone.properties
    #----------------------
    #修改下面的几个参数，其他的不变
    bootstrap.servers=wykd:9092
    plugin.path=/opt/app/kafka_2.11-2.4.1/lib_conn  #指向刚才放置debezium插件的目录
    
    #这两个参数表示connect消息格式，可选json或avro，在消费的时候也需要使用对应的格式，本次我们使用json格式
    key.converter=org.apache.kafka.connect.json.JsonConverter
    value.converter=org.apache.kafka.connect.json.JsonConverter

**3. 配置debezium mysql connector**

创建debezium mysql connector插件配置文件，核心参数如下，其他参数请参考官方手册：[mysql-connector-configuration-properties\_debezium][mysql-connector-configuration-properties_debezium]

#修改kafka connect 配置文件
    vim /opt/app/kafka/connect-debezium-mysql-wyk-csdn.properties
    #----------------------
    name=wyk_csdn_mysql_debezium_connector    #connector 名称，唯一
    connector.class=io.debezium.connector.mysql.MySqlConnector  #connector 插件
    database.hostname=wykd        #数据库host
    database.port=3306            #数据库port
    database.user=root            #数据库user 有复制binlog权限
    database.password=123456      #数据库password
    database.server.id=9999       # server id 同一个数据库的serverId不能有重复的
    database.server.name=wyk_sandbox    # server name [记录DDL动作的topic名称，给消费端使用的]
    database.whitelist=testschema       # schema白名单，只监控白名单库下的，和blacklist参数互相冲突
    database.history.kafka.bootstrap.servers=wykd:9092  # kafka
    database.history.kafka.topic=wyk_csdn_debezium_mysql  #记录所有schema变动记录的topic[给连接器使用的]
    include.schema.changes=true  # 监控schema变动，写入一个名称和server name一样的topic内
    include.query=true        # 将执行的sql也记录下来，需要mysql端的参数binlog_rows_query_log_events=ON

> **注意：**
> 
> 配置文件中每行参数前后不能有空格；
> 
> 如果需要在消息内记录执行的sql，需要：
> 
>  *  mysql：set global binlog\_rows\_query\_log\_events=ON;
>  *  debezium mysql connect配置文件：include.query=true
> 
> 需要特别关注的几个参数：whitelist、blacklist、include.query、include.schema.changes、snapshot.mode

# 注册启动Connector #

如果当前kafka没有启动kafka connect服务，则使用下面的命令启动并注册connect到kafka：

cd $KAFKA_HOME
    bin/connect-standalone.sh config/connect-standalone.properties config/connect-debezium-mysql-wyk-csdn2.properties

启动成功后，会一并启动kafka rest服务，我们可以使用restful api的方式查看或管理connectors：

GET /Connectors：返回活跃的 Connector 列表
    POST /Connectors：创建一个新的 Connector；请求的主体是一个包含字符串name字段和对象 config 字段（Connector 的配置参数）的 JSON 对象。
    GET /Connectors/{name}：获取指定 Connector 的信息
    GET /Connectors/{name}/config：获取指定 Connector 的配置参数
    PUT /Connectors/{name}/config：更新指定 Connector 的配置参数
    GET /Connectors/{name}/status：获取 Connector 的当前状态，包括它是否正在运行，失败，暂停等。
    GET /Connectors/{name}/tasks：获取当前正在运行的 Connector 的任务列表。
    GET /Connectors/{name}/tasks/{taskid}/status：获取任务的当前状态，包括是否是运行中的，失败的，暂停的等，
    PUT /Connectors/{name}/pause：暂停连接器和它的任务，停止消息处理，直到 Connector 恢复。
    PUT /Connectors/{name}/resume：恢复暂停的 Connector（如果 Connector 没有暂停，则什么都不做）
    POST /Connectors/{name}/restart：重启 Connector（Connector 已故障）
    POST /Connectors/{name}/tasks/{taskId}/restart：重启单个任务 (通常这个任务已失败)
    DELETE /Connectors/{name}：删除 Connector, 停止所有的任务并删除其配置

在浏览器查看当前活跃的connectors： *ip:8083/connectors*

![20200925170945903.png][]

例：此时我们想再注册一个debezium mysql connector服务的话就不能用刚刚的命令启动了，会报错提示Failed to bind to 0.0.0.0/0.0.0.0:8083，Caused by: java.net.BindException: Address already in use。这里的8083端口就是kafka connect服务，因此我们只能使用下面的命令来注册新的connector：

# 这里的 connector 配置需要使用json格式
    curl -s http://10.1.174.10:8083/connectors -X POST -H "Content-Type: application/json" --data \
    '{"name": "wyk_csdn_mysql_debezium_connector2","config": {"connector.class": "io.debezium.connector.mysql.MySqlConnector", "database.hostname": "10.1.174.10", "database.port": "3306", "database.user": "root", "database.password": "123456", "database.server.id": "9988", "database.server.name": "wyk_sandbox", "database.whitelist": "canal_manager", "database.history.kafka.bootstrap.servers": "10.1.174.10:9092 ", "database.history.kafka.topic": "wyk_csdn_debezium_mysql2", "include.schema.changes": "true","include.query":"true" }}'

例：使用rest api管理kafka connectors：

# 查看所有活跃的connectors
    curl -s http://10.1.174.10:8083/connectors -X GET
    
    # 注册一个新的connector
    curl -s http://10.1.174.10:8083/connectors -X POST -H "Content-Type: application/json" --data \
    '{"name": "wyk_csdn_mysql_debezium_connector2","config": {"connector.class": "io.debezium.connector.mysql.MySqlConnector", "database.hostname": "10.1.174.10", "database.port": "3306", "database.user": "root", "database.password": "123456", "database.server.id": "9988", "database.server.name": "wyk_sandbox", "database.whitelist": "canal_manager", "database.history.kafka.bootstrap.servers": "10.1.174.10:9092 ", "database.history.kafka.topic": "wyk_csdn_debezium_mysql", "include.schema.changes": "true","include.query":"true" }}'
    
    # 删除指定的connector
    curl -s http://10.1.174.10:8083/connectors/wyk_csdn_mysql_debezium_connector2 -X DELETE

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 1][]

# 验证CDC #

此时我们已经启动了debezium mysql connect来监控指定的mysql并将数据写入了kafka，和maxwell,canal不同的是，debezium监控的消息会将每张表对应一个topic，且该topic只包含该表的增删改(insert/delete/update)操作，DDL操作如create/drop/truncate/rename/alter等命令会统一写入以配置文件中的*database.server.name* 参数的值为名称的topic内。

**Topic 命名方式**：对于同一个connector 监控的数据范围，一张表对应一个topic，该范围内所有库所有表的DDL操作记录在同一个topic内：

*  **DDL topic**：serverName
 *  **DML topic**：serverName.databaseName.tableName

![20200925172137943.png][]

默认情况下，为了提升性能及保证数据一致性，在第一次启动时并不会从binlog回放表的每个操作来同步数据，而是通过一个全局读锁将表的快照直接同步到对应的topic内，这个过程很快，在同步完成后才会继续监控binlog内的操作达到初始化时全量切实时的目的。该**快照模式**的参数在connector配置文件中为 ***snapshot.mode***，有下面几个可选的值：

*  **initial**：默认值，第一次启动connector时同步数据快照，完成后再从记录的binlog位置开始同步，但当关闭connector很久以后再次启动时若上次记录的Offset在binlog内已经被删除的时候就会启动失败（建议mysql保留全部binlog）；
 *  **when\_needed**：在有必要的任何时候执行快照任务，和initial类似，不同的是当遇到故障时如binlog丢失时会重新发起一个快照同步；
 *  **never**：永不执行快照同步，从Binlog记录的位置回放；
 *  **schema\_only**：从第一次启动后记录当前schema，然后从binlog位置开始同步，仅同步schema变动而不同步数据；
 *  **schema\_only\_recovery**：仅同步schema变更而不同步数据，只能配置于已有的connector之上，不是从当前启动时记录schema而是从已存在的connector上次记录的位置开始；

**消息体： **分为两种，一个记录DDL，一个记录DML

DDL：[schema\_change\_topic\_structure][schema_change_topic_structure]

DML：[change\_event\_values][change_event_values] ，默认会使用primary key 或unique key作为消息体的key，也可自定义key。

> DML中insert对应的op为**c**，update对应为**u**，delete对应为**d**。

**例：**

DDL：

{
    	"schema": {
    		"type": "struct",
    		"fields": [
    			{
    				"type": "struct",
    				"fields": [
    					{
    						"type": "string",
    						"optional": false,
    						"field": "version"
    					},
    					{
    						"type": "string",
    						"optional": false,
    						"field": "connector"
    					},
    					{
    						"type": "string",
    						"optional": false,
    						"field": "name"
    					},
    					{
    						"type": "int64",
    						"optional": false,
    						"field": "ts_ms"
    					},
    					{
    						"type": "string",
    						"optional": true,
    						"name": "io.debezium.data.Enum",
    						"version": 1,
    						"parameters": {
    							"allowed": "true,last,false"
    						},
    						"default": "false",
    						"field": "snapshot"
    					},
    					{
    						"type": "string",
    						"optional": false,
    						"field": "db"
    					},
    					{
    						"type": "string",
    						"optional": true,
    						"field": "table"
    					},
    					{
    						"type": "int64",
    						"optional": false,
    						"field": "server_id"
    					},
    					{
    						"type": "string",
    						"optional": true,
    						"field": "gtid"
    					},
    					{
    						"type": "string",
    						"optional": false,
    						"field": "file"
    					},
    					{
    						"type": "int64",
    						"optional": false,
    						"field": "pos"
    					},
    					{
    						"type": "int32",
    						"optional": false,
    						"field": "row"
    					},
    					{
    						"type": "int64",
    						"optional": true,
    						"field": "thread"
    					},
    					{
    						"type": "string",
    						"optional": true,
    						"field": "query"
    					}
    				],
    				"optional": false,
    				"name": "io.debezium.connector.mysql.Source",
    				"field": "source"
    			},
    			{
    				"type": "string",
    				"optional": false,
    				"field": "databaseName"
    			},
    			{
    				"type": "string",
    				"optional": false,
    				"field": "ddl"
    			}
    		],
    		"optional": false,
    		"name": "io.debezium.connector.mysql.SchemaChangeValue"
    	},
    	"payload": {
    		"source": {
    			"version": "1.2.2.Final",
    			"connector": "mysql",
    			"name": "wyk_sandbox",
    			"ts_ms": 1601016725000,
    			"snapshot": "false",
    			"db": "testschema",
    			"table": "wyk_csdn",
    			"server_id": 1,
    			"gtid": null,
    			"file": "mysql-bin.000016",
    			"pos": 93212,
    			"row": 0,
    			"thread": null,
    			"query": null
    		},
    		"databaseName": "testschema",
    		"ddl": "create table wyk_csdn(id int,name varchar(20),ins_ts timestamp)"
    	}
    }

DML：

{
    	"schema": {
    		"type": "struct",
    		"fields": [
    			{
    				"type": "struct",
    				"fields": [
    					{
    						"type": "int32",
    						"optional": true,
    						"field": "id"
    					},
    					{
    						"type": "string",
    						"optional": true,
    						"field": "name"
    					},
    					{
    						"type": "string",
    						"optional": true,
    						"name": "io.debezium.time.ZonedTimestamp",
    						"version": 1,
    						"field": "ins_ts"
    					}
    				],
    				"optional": true,
    				"name": "wyk_sandbox.testschema.wyk_csdn.Value",
    				"field": "before"
    			},
    			{
    				"type": "struct",
    				"fields": [
    					{
    						"type": "int32",
    						"optional": true,
    						"field": "id"
    					},
    					{
    						"type": "string",
    						"optional": true,
    						"field": "name"
    					},
    					{
    						"type": "string",
    						"optional": true,
    						"name": "io.debezium.time.ZonedTimestamp",
    						"version": 1,
    						"field": "ins_ts"
    					}
    				],
    				"optional": true,
    				"name": "wyk_sandbox.testschema.wyk_csdn.Value",
    				"field": "after"
    			},
    			{
    				"type": "struct",
    				"fields": [
    					{
    						"type": "string",
    						"optional": false,
    						"field": "version"
    					},
    					{
    						"type": "string",
    						"optional": false,
    						"field": "connector"
    					},
    					{
    						"type": "string",
    						"optional": false,
    						"field": "name"
    					},
    					{
    						"type": "int64",
    						"optional": false,
    						"field": "ts_ms"
    					},
    					{
    						"type": "string",
    						"optional": true,
    						"name": "io.debezium.data.Enum",
    						"version": 1,
    						"parameters": {
    							"allowed": "true,last,false"
    						},
    						"default": "false",
    						"field": "snapshot"
    					},
    					{
    						"type": "string",
    						"optional": false,
    						"field": "db"
    					},
    					{
    						"type": "string",
    						"optional": true,
    						"field": "table"
    					},
    					{
    						"type": "int64",
    						"optional": false,
    						"field": "server_id"
    					},
    					{
    						"type": "string",
    						"optional": true,
    						"field": "gtid"
    					},
    					{
    						"type": "string",
    						"optional": false,
    						"field": "file"
    					},
    					{
    						"type": "int64",
    						"optional": false,
    						"field": "pos"
    					},
    					{
    						"type": "int32",
    						"optional": false,
    						"field": "row"
    					},
    					{
    						"type": "int64",
    						"optional": true,
    						"field": "thread"
    					},
    					{
    						"type": "string",
    						"optional": true,
    						"field": "query"
    					}
    				],
    				"optional": false,
    				"name": "io.debezium.connector.mysql.Source",
    				"field": "source"
    			},
    			{
    				"type": "string",
    				"optional": false,
    				"field": "op"
    			},
    			{
    				"type": "int64",
    				"optional": true,
    				"field": "ts_ms"
    			},
    			{
    				"type": "struct",
    				"fields": [
    					{
    						"type": "string",
    						"optional": false,
    						"field": "id"
    					},
    					{
    						"type": "int64",
    						"optional": false,
    						"field": "total_order"
    					},
    					{
    						"type": "int64",
    						"optional": false,
    						"field": "data_collection_order"
    					}
    				],
    				"optional": true,
    				"field": "transaction"
    			}
    		],
    		"optional": false,
    		"name": "wyk_sandbox.testschema.wyk_csdn.Envelope"
    	},
    	"payload": {
    		"before": null,
    		"after": {
    			"id": 2,
    			"name": "wyk2",
    			"ins_ts": "2020-09-25T08:10:22Z"
    		},
    		"source": {
    			"version": "1.2.2.Final",
    			"connector": "mysql",
    			"name": "wyk_sandbox",
    			"ts_ms": 1601021422000,
    			"snapshot": "false",
    			"db": "testschema",
    			"table": "wyk_csdn",
    			"server_id": 1,
    			"gtid": null,
    			"file": "mysql-bin.000016",
    			"pos": 97798,
    			"row": 0,
    			"thread": 162,
    			"query": "insert into wyk_csdn values(2,'wyk2',current_timestamp())"
    		},
    		"op": "c",
    		"ts_ms": 1601021422679,
    		"transaction": null
    	}
    }

另外如果使用confluent platform，可以在control center中直接查看管理connectors以及查看topic内的数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 2][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 4][]

# 尾巴 #

**优点：**

1.  debezium支持全DDL（create/alter/rename/drop/truncate）和DML（insert/update/delete）的操作日志同步；（Maxwell不支持truncate命令同步）
2.  支持使用数据的主键或唯一键作为消息体的key（可以根据该唯一键做Topic分区进行负载均衡）；
3.  支持快照模式(snapshot.mode)全量同步；（Canal不支持全量，Maxwell支持bootstrap全量）
4.  还支持多种数据源（MySQL，Oracle，Postgre，MongoDB，SqlServer，DB2，Cassandra），比Canal，Maxwell功能更强大；
5.  社区活跃，与kafka connect配合使用如虎添翼。

**缺点：**

1.  需要kafka connect依赖；
2.  消息体内容太多，对消息队列压力较大；（可使用transforms 模块进行压缩和定制）
3.  每张表对应一个topic，管理起来不够方便（canal，maxwell则可以使用正则来处理）； （可使用transforms 模块进行压缩和定制）

希望本文对你有帮助，请点个赞鼓励一下作者吧~ 谢谢！

[CDC_Canal _WebUI]: https://blog.csdn.net/wsdc0521/article/details/108727632
[CDC_Maxwell_v1.27.1 _MySQL_Kafka]: https://blog.csdn.net/wsdc0521/article/details/108736260
[kafka 2.4.1]: https://blog.csdn.net/wsdc0521/article/details/105813493
[Kafka_2.6.0_kafka]: https://blog.csdn.net/wsdc0521/article/details/108597755
[Kafka_Kafka_ Confluent Platform]: https://blog.csdn.net/wsdc0521/article/details/108764050
[https_debezium.io_releases_1.2]: https://debezium.io/releases/1.2/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70]: /images/20221123/5a7c6d1ae4af488f9c09c6c05a787c00.png
[mysql-connector-configuration-properties_debezium]: https://debezium.io/documentation/reference/1.2/connectors/mysql.html#mysql-connector-configuration-properties_debezium
[20200925170945903.png]: /images/20221123/ea938912942e4c1ab8fe7319167cc48a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 1]: /images/20221123/a322f69d030149c899e1d8d33b2456be.png
[20200925172137943.png]: /images/20221123/361e2c1da2624db59898ba453618af5b.png
[schema_change_topic_structure]: https://debezium.io/documentation/reference/1.2/connectors/mysql.html#_schema_change_topic_structure
[change_event_values]: https://debezium.io/documentation/reference/1.2/connectors/mysql.html#_change_event_values
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 2]: /images/20221123/3642d2946e084533b72e5abc3c17bdbc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 3]: /images/20221123/87d4dc495988444694311e138d362767.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGMwNTIx_size_16_color_FFFFFF_t_70 4]: /images/20221123/302fc835160649b0aea668184833a9be.png