Debezium源连接器配置：跳过全量快照 仅捕获增量变更记录

针对你需要仅捕获连接器启动后的增量变更、跳过默认全量数据同步的场景，通过调整连接器的快照相关配置即可实现，不需要改动现有同步链路的核心逻辑。

核心配置调整项 #

你当前遇到的首次启动拉取全量数据的问题，根源是Debezium MySQL连接器默认采用initial快照模式：启动时会先将所有白名单范围内的表存量数据全量同步到Kafka，完成后再从快照对应的binlog位点开始捕获增量。需要做的调整如下：

• 新增核心配置参数snapshot.mode，取值设置为no-data

  该模式的执行逻辑为：启动时仅收集白名单范围内表的结构元数据用于后续binlog解析校验，不导出任何存量业务数据，直接从当前数据库的最新binlog位点开始捕获后续所有变更，完全匹配你“两个库存量完全一致、仅需同步启动后增量”的场景
  注意不要误用never模式：该模式会跳过所有快照步骤（包括结构元数据收集），后续遇到表结构变更时极易出现binlog解析错位、同步数据错乱的问题

• 检查database.history.kafka.topic的配置值
  如果你是首次部署该同步链路，不要沿用已有历史数据的旧topic：Debezium会将快照元数据、binlog位点记录保存在这个topic中，启动时如果检测到该topic里已有历史保存的快照位点记录，会自动从记录的旧位点开始恢复，甚至触发全量快照逻辑，导致跳过全量的配置失效。首次部署建议使用全新topic名称；如果是调整现有运行中的配置，需要先清空该topic的存量数据再重启连接器
• 修正原有配置的语法错误
  你当前配置中transforms字段声明了11个转换规则，但仅配置了unwrap和dropTopicPrefix的具体实现类，其余规则没有对应参数会导致连接器启动失败。需要将未定义的转换规则从列表中删除，或者补充对应转换的实现配置。

调整后的完整连接器配置示例 #

对比你原来的配置，仅做两处核心改动：新增snapshot.mode: no-data参数、删除transforms列表中未定义的无效规则，完整配置如下：

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '
{
    "name": "mysql5-source",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "10",
        "database.hostname": "host",
        "database.port": "3307",
        "database.user": "root",
        "database.password": "secret",
        "database.server.id": "11",
        "database.server.name": "dbserver",
        "database.whitelist": "dbname",
        "table.whitelist": "dbname.exm1,dbname.exm4,dbname.exm2,dbname.exm3",
        "snapshot.mode": "no-data",
        "database.history.kafka.bootstrap.servers": "kafka:29092",
        "database.history.kafka.topic": "mysql5table",
        "key.converter": "io.confluent.connect.avro.AvroConverter",
        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "key.converter.schema.registry.url": "http://schema-registry:8081",
        "value.converter.schema.registry.url": "http://schema-registry:8081",
        "key.converter.schemas.enable": true,
        "value.converter.schemas.enable": true,
        "transforms": "unwrap,dropTopicPrefix",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "transforms.dropTopicPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.dropTopicPrefix.regex": "dbserver.indianmo_imc_new.(.*)",
        "transforms.dropTopicPrefix.replacement": "$1"
  }
}'

上线验证要点 #

• 连接器启动后，第一时间观察对应业务表的Kafka Topic消息量，确认没有批量的存量数据同步消息
• 在db1端触发一条测试数据的新增/更新操作，确认变更可以被正常捕获并投递到Kafka
• 确认源端捕获逻辑正常后，再对接db2的sink连接器完成增量写入即可

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