debezium 小试
2530 字
13 分钟
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
- 墨滴社区
文章分享
如果这篇文章对你有帮助，欢迎分享给更多人！
相关文章智能推荐
1
在 PVE 上给软路由虚拟机加一个网络 Watchdog
技术分享在 PVE 宿主机上用 Bash 和 systemd timer 实现一个网络 watchdog：先检测公网连通性，连续失败后重启软路由虚拟机，多次自愈无效再升级为重启 PVE 宿主机。
2
命令行自动登录 JumpServer：SSH、TOTP 与密码提示的自动化
技术分享把 JumpServer 登录流程拆成凭据读取和终端交互两层：zsh 从密码管理器安全读取 TOTP seed 与可选密码，expect 只负责响应 SSH 登录过程中的 password 和 Code 提示。
3
git 个人开发工作流
技术分享介绍个人 Git 分支协作流程：以基线分支（如 feature-a）+ 个人开发分支（feature-a-xxx）进行开发，开发分支定期 rebase 同步上游，最终通过 merge（必要时先 squash 整理提交）合回基线；并提供 grebase/gmergebase 等脚本，支持自动推导基线分支、预览计划、带 autostash 的安全 rebase，以及可选 squash/推送/删除分支的安全 merge-back。
4
使用 docker-atrust 连入深信服
技术分享使用 Docker 容器运行深信服 aTrust 客户端，通过 VNC 完成登录，并结合本机 mihomo 配置实现公司内网的 DNS 解析与按域名、IP 的精确代理分流；宿主机暴露 1080 与 8888 端口供 socks5 和 HTTP 代理使用，从而在不安装原生客户端的情况下实现零信任接入并降低对本机的影响。
5
mihomo 使用
技术分享2025-09-30
随机文章随机推荐
文章
61
分类
4
标签
52
总字数
64,663
运行时长
0 天
最后活动
0 天前
站点域名blog.sephy.top
Fireflyv6.13.9
Astrov7.0.7
Nodev22.23.1
pnpmv9.14.4
构建时间2026年7月15日 00:54:40
系统信息Linux / x86_64