Debezium与Kafka Connect：能否仅追踪SQL筛选的表子集变更？

结论 #

完全可以实现，但transforms.filter确实不支持你这种带子查询的复杂SQL过滤逻辑，得换其他方式来做。

可行的几种方案 #

1. 数据库建视图，让Debezium捕获视图变更 #

• 先在数据库里建一个符合你过滤条件的视图：

  CREATE VIEW filtered_some_table AS
  SELECT * FROM some_table 
  WHERE some_fk_id NOT IN (
      SELECT some_fk_id FROM some_table_2 WHERE some_field <> 'some_value'
  );
  

• 然后配置Debezium去捕获这个视图的变更就行。不过要注意，不同数据库对视图CDC的支持不一样：比如PostgreSQL得确保视图是可更新的，或者逻辑复制得开视图捕获；MySQL如果视图是多表关联的话可能受限，具体得看你用的数据库的CDC文档。

2. 自定义Debezium消息转换器（SMT） #

• 要是需要更灵活的逻辑，可以自己写个SMT来实现这个过滤：
  • 用Java实现io.debezium.transforms.spi.Transformation接口
  • 在转换逻辑里，拿变更事件里的some_fk_id去查数据库（最好加缓存，不然频繁查库影响性能），判断是否符合过滤条件
  • 配置Debezium用你这个自定义SMT，示例配置大概是这样：

    transforms=customFilter
    transforms.customFilter.type=com.yourcompany.transforms.CustomFilterTransform
    transforms.customFilter.db.url=jdbc:mysql://localhost:3306/your_db
    transforms.customFilter.db.user=db_user
    transforms.customFilter.db.password=db_pass
    

3. 用触发器+中间表间接实现 #

• 在some_table和some_table_2上建触发器，当数据变的时候，把符合过滤条件的记录同步到一张中间表
• 然后让Debezium只抓这个中间表的变更
• 这个方法要注意，触发器会给数据库加性能开销，还要考虑数据一致性的问题

关于transforms.filter的补充（原文档翻译） #

Debezium的transforms.filter只支持基于事件内容的简单判断，比如匹配字段值、比较大小这些，用的是Apache Commons JEXL表达式，完全不支持SQL查询或者跨表关联的复杂逻辑。

给个简单的使用例子参考：

transforms=filter
transforms.filter.type=io.debezium.transforms.Filter
transforms.filter.language=jexl
transforms.filter.condition=payload.after.some_fk_id == '123'

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