如何配置Debezium仅在column.include.list列变更时向Kafka发消息？

问题根源 #

column.include.list仅负责裁剪最终发送到Kafka的消息字段，只保留你指定的列，但Debezium是基于数据库binlog触发事件的——只要表有任何变更（哪怕只修改了不在白名单里的列），它都会生成事件，只是把非白名单列从before/after中剔除，所以你会收到那些before和after完全一致的无意义消息。

最简便的方案：MySQL专属include.query.filter配置 #

如果你使用的是MySQL连接器，直接在源端配置过滤规则，从根上跳过仅修改非关注列的事件，比SMT过滤效率更高（无需生成消息再丢弃）。

配置示例 #

假设你要监控的列是id、name、email，添加以下配置：

include.query.filter=ALTER TABLE.* CHANGE.*(id|name|email)|INSERT.*(id|name|email)|UPDATE.*SET.*(id|name|email)

这个正则会匹配所有修改指定列的INSERT/UPDATE操作，以及涉及这些列的ALTER TABLE语句，只有匹配到的操作才会被Debezium捕获处理。请根据实际列名调整正则中的列部分。

跨数据库兼容方案：SMT Filtering #

如果需要支持多数据库（如同时用MySQL和PostgreSQL），则使用Debezium的Filter转换，在消息生成后过滤掉before/after无差异的消息。

配置示例 #

transforms=filterUnchanged
transforms.filterUnchanged.type=io.debezium.transforms.Filter
transforms.filterUnchanged.language=jsr223.groovy
transforms.filterUnchanged.condition=
    def targetCols = ['id', 'name', 'email']
    targetCols.any { col -> record.before[col] != record.after[col] }

该Groovy脚本会检查指定列中是否有任意一列的值发生变化，仅当存在变化时才保留消息，否则直接丢弃。

注意事项 #

• include.query.filter是MySQL连接器独有的配置，依赖对binlog中SQL语句的解析，需确保MySQL binlog格式符合Debezium要求（默认行模式下该配置依然有效）。
• 正则需覆盖所有可能修改目标列的SQL写法，建议测试常见修改场景，确保正则匹配准确。
• SMT方案兼容性好，但会在消息生成后过滤，性能略低于源端过滤，适合跨数据库场景。

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