如何配置Debezium MySQL Kafka Connect仅捕获新变更日志数据？

要避免Debezium同步表中旧记录，只捕获后续的新变更，只需调整连接器的快照模式配置即可：

核心配置修改 #

在你的Kafka Connect配置中添加以下参数：

"snapshot.mode": "schema_only"

参数说明 #

• schema_only：连接器仅同步目标表的结构（Schema），不会对现有数据执行全量快照，之后仅捕获表的新增/修改/删除等变更操作，完全匹配你的需求。
• 若你确认无需同步表结构（如已提前在Schema Registry注册过对应Schema），也可使用snapshot.mode: never，但不推荐此方式——缺少Schema同步可能导致后续CDC数据解析异常。

修改后的完整配置示例 #

{
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "database.hostname": "mysql",
        "database.port": "3306",
        "database.user": "xxxx",
        "database.password": "xxxx",
        "database.server.id": "42",
        "database.server.name": "xxxx",
        "table.whitelist": "demo.movies",
        "database.history.kafka.bootstrap.servers": "broker:9092",
        "database.history.kafka.topic": "dbhistory.demo" ,
        "decimal.handling.mode": "double",
        "include.schema.changes": "true",
        "transforms": "unwrap,dropTopicPrefix",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "transforms.dropTopicPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.dropTopicPrefix.regex":"asgard.demo.(.*)",
        "transforms.dropTopicPrefix.replacement":"$1",
        "key.converter": "io.confluent.connect.avro.AvroConverter",
        "key.converter.schema.registry.url": "http://schema-registry:8081",
        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "value.converter.schema.registry.url": "http://schema-registry:8081",
        "snapshot.mode": "schema_only"
}

注意事项 #

1. 若该连接器已运行过并生成全量快照数据，修改配置后重启前，建议清理对应的database.history.kafka.topic（即dbhistory.demo），避免连接器复用之前的快照状态。
2. 确保MySQL的binlog已开启且配置正确，Debezium依赖binlog捕获后续变更操作。

内容的提问来源于stack exchange，提问作者Forece85