InterestingLab/Waterdrop项目中Postgres-CDC数据重复问题解析与解决方案

背景介绍

在使用InterestingLab/Waterdrop项目进行数据同步时，当配置Postgres-CDC作为数据源，同时使用RabbitMQ和Console作为数据接收端时，出现了数据重复的问题。这个问题特别出现在处理CDC(变更数据捕获)源中的更新记录时，值得深入分析其产生原因和解决方案。

问题现象

当Postgres数据库中的表发生更新操作时，Waterdrop会捕获到两条记录：

1. UPDATE_BEFORE - 表示更新前的数据状态
2. UPDATE_AFTER - 表示更新后的数据状态

在数据接收端(RabbitMQ和Console)会同时收到这两条记录，导致数据重复。而在JDBC接收端则不会出现此问题，因为JDBC连接器内部已经实现了对UPDATE_BEFORE记录的过滤逻辑。

技术原理分析

Postgres-CDC的工作原理是基于PostgreSQL的逻辑解码功能，它会捕获数据库中的所有数据变更事件，包括：

• 插入(INSERT)
• 更新(UPDATE)
• 删除(DELETE)

对于UPDATE操作，CDC会生成两条记录：

1. 更新前的数据快照(UPDATE_BEFORE)
2. 更新后的数据快照(UPDATE_AFTER)

这种设计是为了完整记录数据变更的历史，但在某些业务场景下，我们可能只需要关注变更后的数据状态。

解决方案

方案一：使用FilterRowKind转换器

Waterdrop提供了FilterRowKind转换器，可以过滤掉不需要的变更类型记录。配置示例如下：

transform {
  FilterRowKind {
    source_table_name = "employees"
    result_table_name = "employees_filtered"
    exclude_kinds = ["UPDATE_BEFORE"]
  }
}

这种方案的优势是：

1. 配置简单直观
2. 可以灵活选择需要保留的变更类型
3. 不影响其他接收端的处理逻辑

方案二：在接收端实现过滤逻辑

类似JDBC接收端的做法，可以在自定义接收器中实现对特定变更类型的过滤。这种方法需要一定的开发工作，但可以实现更精细的控制。

最佳实践建议

1. 根据业务需求明确需要处理的变更类型
2. 对于只需要最终状态的场景，建议过滤掉UPDATE_BEFORE记录
3. 对于需要完整变更历史的场景，可以考虑将记录类型作为元数据一并存储
4. 在测试环境中验证过滤逻辑是否符合预期

总结

Postgres-CDC产生的数据重复问题本质上是由于其完整记录变更历史的特性导致的。通过合理使用Waterdrop提供的转换器，可以灵活地控制数据处理流程，满足不同业务场景的需求。理解CDC的工作原理和Waterdrop的转换机制，有助于构建更健壮的数据管道。