debezium 小试
以前用过 maxwell,几年前就看到过 debezium,近期调研替换阿里云 DTS 订阅的方案,尝试一下
安装步骤
使用 docker compose 方式安装,相关配置如下
service: # https://hub.docker.com/r/bitnami/kafka kafka: image: bitnami/kafka:3.4.1 container_name: kafka volumes: - $HOME/data/kafka/:/bitnami/kafka/ #注意目录权限问题 environment: - KAFKA_CFG_NODE_ID=0 - KAFKA_CFG_PROCESS_ROLES=controller,broker - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER # EXTERNAL 配置的是挡墙 Docker 所在的主机地址 - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094 - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,EXTERNAL://192.168.50.21:9094 # 宿主机ip - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT ports: - 9092:9092 - 9093:9093 - 9094:9094
debezium-connect: image: debezium/connect:3.0.0.Final container_name: debezium-connect environment: - TZ=Asia/Shanghai - GROUP_ID=debezium - CONFIG_STORAGE_TOPIC=debezium_config_storage - OFFSET_STORAGE_TOPIC=debezium_offset_storage - BOOTSTRAP_SERVERS=kafka:9094 ports: - 8083:8083 depends_on: - kafka启动
docker compose --compatibility up -d debezium-connect准备 mysql 复制账号,这里用户名密码都使用 slave_debezium,并给予从库权限
CREATE USER 'slave_debezium'@'%' IDENTIFIED BY 'slave_debezium';GRANT REPLICATION SLAVE ON *.* TO 'slave_debezium'@'%';FLUSH PRIVILEGES;确认 mysql 主从状态
# show master status; # 旧语法show binary log status; # mysql 8.4 的语法创建 debezium 信号表
CREATE TABLE `debezium_signal` ( `id` varchar(64) NOT NULL, `type` varchar(32) NOT NULL, `data` json DEFAULT NULL, PRIMARY KEY (`id`));初始化 debezium-connect
关键配置说明
- table.include.list 要监听的对象表(命名形式 库名.表名,库名.表名)
- snapshot.mode = schema_only (在任务启动时,只做表结构快照,不做数据快照,默认的 initial 策略会对监听的表做全量快照,遇到数据量大的表可能会影响生产写入)
- snapshot.locking.mode = none(快照时不使用锁,避免影响生产写操作)
- signal.data.collection = app_dev.debezium_signal(指定 dbz 的 信号表,用于触发增量快照)
- read.only = true (只读账号时必须指定,否则会尝试对 debezium_signal 表做 insert 操作来产生水位信息)
- transforms = reroute (SMT 相关配置,通过正则表达式调整 dbz 默认的 topic 策略)
- decimal.handling.mode = string (将 mysql 中 decimal 类型转换为 string 输出, 确保不丢失精度)
- converters = bitOneToInt (默认 bit 会转换为 boolean 类型,这里通过自定义 CustomConvert 修改默认行为)下面暂时没用到
# 创建新的 connectorcurl -i -X POST http://192.168.50.21:8083/connectors \ -H "Accept:application/json" \ -H "Content-Type:application/json" \ -d \ '{ "name": "debezium-connector-mysql", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "database.hostname": "192.168.50.21", "database.port": "3306", "database.user": "slave_debezium", "database.password": "slave_debezium", "database.server.id": "101", "topic.prefix": "debeziumdemo", "table.include.list": "app_dev.cdc_demo", "schema.history.internal.kafka.bootstrap.servers": "192.168.50.21:9094", "schema.history.internal.kafka.topic": "debezium_schema_history", "include.schema.changes": "true", "snapshot.mode": "schema_only", "snapshot.locking.mode": "none", "signal.data.collection": "app_dev.debezium_signal", "read.only": true, "transforms": "reroute", "transforms.reroute.type": "io.debezium.transforms.ByLogicalTableRouter", "transforms.reroute.topic.regex": "(\\S+)\\.(\\S+)\\.(\\S+)", "transforms.retopic.key.enforce.uniqueness": "false", "transforms.reroute.topic.replacement": "$1.$2.debezium_cdc", "time.precision.mode": "adaptive_time_microseconds", "decimal.handling.mode": "string", "boolean.length.checker": "false" } }'通过 Offset Explorer 3 可以看到相关数据
![[image 29.png]]
debezium-connect 常用 rest 操作
获取 connect 状态
# 获取状态curl -X GET http://host:port/connectors/debezium-connector-mysql/status获取 connect 配置
curl -X GET http://host:port/connectors/debezium-connector-mysql/config暂停 connect cdc 捕捉任务
curl -XPUT http://host:port/connectors/debezium-connector-mysql/pause恢复 connect cdc 捕捉任务
curl -XPUT http://host:port/connectors/debezium-connector-mysql/resume插入 debezium_signal 表触发指定表增量快照
INSERT INTO `debezium_signal` (`id`, `type`, `data`) VALUES('ad-hoc-1', 'execute-snapshot', '{\"type\": \"incremental\", \"data-collections\": [\"指定数据库名.指定表名\"]}');问题处理记录
CDC 事件 topic 策略
默认 dbz 默认实现是一个表一个 topic,可以通过 kafka connect 的 SMT 机制来重路由,初始配置如下
"transforms": "reroute","transforms.reroute.type": "io.debezium.transforms.ByLogicalTableRouter","transforms.reroute.topic.regex": "(\\S+)\\.(\\w+)\\.(\\w+)","transforms.retopic.key.enforce.uniqueness": "false","transforms.reroute.topic.replacement": "$1.$2.debezium_cdc",但是查看 kafka 队列发现不生效!
![[image 30.png]]
原因是 正则表达式 \w 不包含中划线,需要修改为 \S
系统提示 debezium_signal 没有写入权限
在做增量快照时,后台报错提示 debezium_signal 没有 insert 权限
解决方案
加上 read.only = true(通过从库账号做只读快照)
MySQL binlog 验证
近期有个疑问,在一个事务开启后,多次对同一条数据进行 update 操作,binlog 会记录几次?
结论 每条 update 操作都会有对应的binlog 日志
验证过程如下
begin;// 第一次修改UPDATE `cdc_demo`SET `col_int`=923,`col_smallint`=302,`col_mediumint`=858,`col_tinyint`=100,`col_bit`=b '1',`col_double`=8.34,`col_float`=567.96,`col_decimal`=734.57,`col_char`='Ri',`col_varchar`='J88fKfgvps',`col_text`='',`col_tinytext`='',`col_mediumtext`='',`col_longtext`='',`col_datetime`='2007-07-28 16:21:13',`col_timestamp`='2020-05-12 06:29:10',`col_date`='2003-03-30',`col_time`='17:33:52',`col_year`=2029 WHERE `col_bigint`=10;// 第二次修改UPDATE `cdc_demo`SET `col_int`=923,`col_smallint`=302,`col_mediumint`=858,`col_tinyint`=100,`col_bit`=b '1',`col_double`=8.34,`col_float`=567.96,`col_decimal`=734.57,`col_char`='Ri',`col_varchar`='J88fKfgvps',`col_text`='',`col_tinytext`='',`col_mediumtext`='',`col_longtext`='',`col_datetime`='2007-07-28 16:21:13',`col_timestamp`='2020-05-12 06:29:10',`col_date`='2003-03-30',`col_time`='17:33:52',`col_year`=2020 WHERE `col_bigint`=10;commit;kafka 中 cdc 的记录,第一次修改
{ "schema": { "type": "struct", "fields": [ { "type": "struct", "fields": [ { "type": "int64", "optional": false, "field": "col_bigint" }, { "type": "int32", "optional": true, "field": "col_int" }, { "type": "int16", "optional": true, "field": "col_smallint" }, { "type": "int32", "optional": true, "field": "col_mediumint" }, { "type": "int16", "optional": true, "field": "col_tinyint" }, { "type": "boolean", "optional": true, "field": "col_bit" }, { "type": "double", "optional": true, "field": "col_double" }, { "type": "float", "optional": true, "field": "col_float" }, { "type": "string", "optional": true, "field": "col_decimal" }, { "type": "string", "optional": true, "field": "col_char" }, { "type": "string", "optional": true, "field": "col_varchar" }, { "type": "string", "optional": true, "field": "col_text" }, { "type": "string", "optional": true, "field": "col_tinytext" }, { "type": "string", "optional": true, "field": "col_mediumtext" }, { "type": "string", "optional": true, "field": "col_longtext" }, { "type": "int64", "optional": true, "name": "io.debezium.time.Timestamp", "version": 1, "field": "col_datetime" }, { "type": "string", "optional": true, "name": "io.debezium.time.ZonedTimestamp", "version": 1, "field": "col_timestamp" }, { "type": "int32", "optional": true, "name": "io.debezium.time.Date", "version": 1, "field": "col_date" }, { "type": "int64", "optional": true, "name": "io.debezium.time.MicroTime", "version": 1, "field": "col_time" }, { "type": "int32", "optional": true, "name": "io.debezium.time.Year", "version": 1, "field": "col_year" } ], "optional": true, "name": "debeziumdemo.app_dev.debezium_cdc.Value", "field": "before" }, { "type": "struct", "fields": [ { "type": "int64", "optional": false, "field": "col_bigint" }, { "type": "int32", "optional": true, "field": "col_int" }, { "type": "int16", "optional": true, "field": "col_smallint" }, { "type": "int32", "optional": true, "field": "col_mediumint" }, { "type": "int16", "optional": true, "field": "col_tinyint" }, { "type": "boolean", "optional": true, "field": "col_bit" }, { "type": "double", "optional": true, "field": "col_double" }, { "type": "float", "optional": true, "field": "col_float" }, { "type": "string", "optional": true, "field": "col_decimal" }, { "type": "string", "optional": true, "field": "col_char" }, { "type": "string", "optional": true, "field": "col_varchar" }, { "type": "string", "optional": true, "field": "col_text" }, { "type": "string", "optional": true, "field": "col_tinytext" }, { "type": "string", "optional": true, "field": "col_mediumtext" }, { "type": "string", "optional": true, "field": "col_longtext" }, { "type": "int64", "optional": true, "name": "io.debezium.time.Timestamp", "version": 1, "field": "col_datetime" }, { "type": "string", "optional": true, "name": "io.debezium.time.ZonedTimestamp", "version": 1, "field": "col_timestamp" }, { "type": "int32", "optional": true, "name": "io.debezium.time.Date", "version": 1, "field": "col_date" }, { "type": "int64", "optional": true, "name": "io.debezium.time.MicroTime", "version": 1, "field": "col_time" }, { "type": "int32", "optional": true, "name": "io.debezium.time.Year", "version": 1, "field": "col_year" } ], "optional": true, "name": "debeziumdemo.app_dev.debezium_cdc.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,incremental" }, "default": "false", "field": "snapshot" }, { "type": "string", "optional": false, "field": "db" }, { "type": "string", "optional": true, "field": "sequence" }, { "type": "int64", "optional": true, "field": "ts_us" }, { "type": "int64", "optional": true, "field": "ts_ns" }, { "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": "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, "name": "event.block", "version": 1, "field": "transaction" }, { "type": "string", "optional": false, "field": "op" }, { "type": "int64", "optional": true, "field": "ts_ms" }, { "type": "int64", "optional": true, "field": "ts_us" }, { "type": "int64", "optional": true, "field": "ts_ns" } ], "optional": false, "name": "debeziumdemo.app_dev.debezium_cdc.Envelope", "version": 2 }, "payload": { "before": { "col_bigint": 10, "col_int": 923, "col_smallint": 302, "col_mediumint": 858, "col_tinyint": 100, "col_bit": true, "col_double": 8.34, "col_float": 567.96, "col_decimal": "734.57", "col_char": "Ri", "col_varchar": "J88fKfgvps", "col_text": "", "col_tinytext": "", "col_mediumtext": "", "col_longtext": "", "col_datetime": 1185639673000, "col_timestamp": "2020-05-11T22:29:10Z", "col_date": 12141, "col_time": 63232000000, "col_year": 2030 }, "after": { "col_bigint": 10, "col_int": 923, "col_smallint": 302, "col_mediumint": 858, "col_tinyint": 100, "col_bit": true, "col_double": 8.34, "col_float": 567.96, "col_decimal": "734.57", "col_char": "Ri", "col_varchar": "J88fKfgvps", "col_text": "", "col_tinytext": "", "col_mediumtext": "", "col_longtext": "", "col_datetime": 1185639673000, "col_timestamp": "2020-05-11T22:29:10Z", "col_date": 12141, "col_time": 63232000000, "col_year": 2029 }, "source": { "version": "3.0.0.Final", "connector": "mysql", "name": "debeziumdemo", "ts_ms": 1742398793844, "snapshot": "false", "db": "app_dev", "sequence": null, "ts_us": 1742398793844717, "ts_ns": 1742398793844717000, "table": "cdc_demo", "server_id": 1, "gtid": "57561b7b-e2a7-11ef-9d3b-0242ac130003:1252076", "file": "mysql-bin.000008", "pos": 819263525, "row": 0, "thread": 165069, "query": null }, "transaction": null, "op": "u", "ts_ms": 1742398793846, "ts_us": 1742398793846407, "ts_ns": 1742398793846407748 }}kafka 中 cdc 的记录,第二次修改
{ "schema": { "type": "struct", "fields": [ { "type": "struct", "fields": [ { "type": "int64", "optional": false, "field": "col_bigint" }, { "type": "int32", "optional": true, "field": "col_int" }, { "type": "int16", "optional": true, "field": "col_smallint" }, { "type": "int32", "optional": true, "field": "col_mediumint" }, { "type": "int16", "optional": true, "field": "col_tinyint" }, { "type": "boolean", "optional": true, "field": "col_bit" }, { "type": "double", "optional": true, "field": "col_double" }, { "type": "float", "optional": true, "field": "col_float" }, { "type": "string", "optional": true, "field": "col_decimal" }, { "type": "string", "optional": true, "field": "col_char" }, { "type": "string", "optional": true, "field": "col_varchar" }, { "type": "string", "optional": true, "field": "col_text" }, { "type": "string", "optional": true, "field": "col_tinytext" }, { "type": "string", "optional": true, "field": "col_mediumtext" }, { "type": "string", "optional": true, "field": "col_longtext" }, { "type": "int64", "optional": true, "name": "io.debezium.time.Timestamp", "version": 1, "field": "col_datetime" }, { "type": "string", "optional": true, "name": "io.debezium.time.ZonedTimestamp", "version": 1, "field": "col_timestamp" }, { "type": "int32", "optional": true, "name": "io.debezium.time.Date", "version": 1, "field": "col_date" }, { "type": "int64", "optional": true, "name": "io.debezium.time.MicroTime", "version": 1, "field": "col_time" }, { "type": "int32", "optional": true, "name": "io.debezium.time.Year", "version": 1, "field": "col_year" } ], "optional": true, "name": "debeziumdemo.app_dev.debezium_cdc.Value", "field": "before" }, { "type": "struct", "fields": [ { "type": "int64", "optional": false, "field": "col_bigint" }, { "type": "int32", "optional": true, "field": "col_int" }, { "type": "int16", "optional": true, "field": "col_smallint" }, { "type": "int32", "optional": true, "field": "col_mediumint" }, { "type": "int16", "optional": true, "field": "col_tinyint" }, { "type": "boolean", "optional": true, "field": "col_bit" }, { "type": "double", "optional": true, "field": "col_double" }, { "type": "float", "optional": true, "field": "col_float" }, { "type": "string", "optional": true, "field": "col_decimal" }, { "type": "string", "optional": true, "field": "col_char" }, { "type": "string", "optional": true, "field": "col_varchar" }, { "type": "string", "optional": true, "field": "col_text" }, { "type": "string", "optional": true, "field": "col_tinytext" }, { "type": "string", "optional": true, "field": "col_mediumtext" }, { "type": "string", "optional": true, "field": "col_longtext" }, { "type": "int64", "optional": true, "name": "io.debezium.time.Timestamp", "version": 1, "field": "col_datetime" }, { "type": "string", "optional": true, "name": "io.debezium.time.ZonedTimestamp", "version": 1, "field": "col_timestamp" }, { "type": "int32", "optional": true, "name": "io.debezium.time.Date", "version": 1, "field": "col_date" }, { "type": "int64", "optional": true, "name": "io.debezium.time.MicroTime", "version": 1, "field": "col_time" }, { "type": "int32", "optional": true, "name": "io.debezium.time.Year", "version": 1, "field": "col_year" } ], "optional": true, "name": "debeziumdemo.app_dev.debezium_cdc.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,incremental" }, "default": "false", "field": "snapshot" }, { "type": "string", "optional": false, "field": "db" }, { "type": "string", "optional": true, "field": "sequence" }, { "type": "int64", "optional": true, "field": "ts_us" }, { "type": "int64", "optional": true, "field": "ts_ns" }, { "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": "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, "name": "event.block", "version": 1, "field": "transaction" }, { "type": "string", "optional": false, "field": "op" }, { "type": "int64", "optional": true, "field": "ts_ms" }, { "type": "int64", "optional": true, "field": "ts_us" }, { "type": "int64", "optional": true, "field": "ts_ns" } ], "optional": false, "name": "debeziumdemo.app_dev.debezium_cdc.Envelope", "version": 2 }, "payload": { "before": { "col_bigint": 10, "col_int": 923, "col_smallint": 302, "col_mediumint": 858, "col_tinyint": 100, "col_bit": true, "col_double": 8.34, "col_float": 567.96, "col_decimal": "734.57", "col_char": "Ri", "col_varchar": "J88fKfgvps", "col_text": "", "col_tinytext": "", "col_mediumtext": "", "col_longtext": "", "col_datetime": 1185639673000, "col_timestamp": "2020-05-11T22:29:10Z", "col_date": 12141, "col_time": 63232000000, "col_year": 2029 }, "after": { "col_bigint": 10, "col_int": 923, "col_smallint": 302, "col_mediumint": 858, "col_tinyint": 100, "col_bit": true, "col_double": 8.34, "col_float": 567.96, "col_decimal": "734.57", "col_char": "Ri", "col_varchar": "J88fKfgvps", "col_text": "", "col_tinytext": "", "col_mediumtext": "", "col_longtext": "", "col_datetime": 1185639673000, "col_timestamp": "2020-05-11T22:29:10Z", "col_date": 12141, "col_time": 63232000000, "col_year": 2020 }, "source": { "version": "3.0.0.Final", "connector": "mysql", "name": "debeziumdemo", "ts_ms": 1742398793844, "snapshot": "false", "db": "app_dev", "sequence": null, "ts_us": 1742398793844717, "ts_ns": 1742398793844717000, "table": "cdc_demo", "server_id": 1, "gtid": "57561b7b-e2a7-11ef-9d3b-0242ac130003:1252076", "file": "mysql-bin.000008", "pos": 819263820, "row": 0, "thread": 165069, "query": null }, "transaction": null, "op": "u", "ts_ms": 1742398793846, "ts_us": 1742398793846719, "ts_ns": 1742398793846719988 }}从上面两条修改记录的 "gtid": "57561b7b-e2a7-11ef-9d3b-0242ac130003:1252076" 可以看出,同一个事务中两次针对同一条数据的 update 操作都会产生对应的 binlog
📎 参考文章
- Read-only Incremental Snapshots for MySQL
- Debezium connector for MySQL :: Debezium Documentation
- Debezium 特性深入介绍 | Amazon Web Services
- Debezium系列- snapshot锁表流程详解_debezium 流程图-CSDN博客
- Fixing Debezium Connectors when they break on production.
- https://medium.com/@a.tambakouzadeh/checklist-to-troubleshoot-kafka-connect-issues-using-debezium-platform-for-cdc-and-mysql-data-b4d517d152a4
- 墨滴社区
文章分享
如果这篇文章对你有帮助,欢迎分享给更多人!













