Apache SeaTunnel 中Postgres-CDC连接器重复数据问题分析与解决方案

问题背景

在使用Apache SeaTunnel 2.3.8版本时，当配置Postgres-CDC作为数据源，同时使用RabbitMQ和Console作为数据接收器(Sink)时，会遇到数据重复的问题。这个问题特别出现在处理CDC(变更数据捕获)源的更新记录时，而同样的配置在使用JDBC作为接收器时则不会出现重复数据。

问题现象

从日志中可以观察到，当Postgres数据库中的表发生更新操作时，CDC源会生成两条记录：

1. 一条带有UPDATE_BEFORE行类型(ROW_KIND=UPDATE_BEFORE)的记录，表示更新前的数据状态
2. 一条带有UPDATE_AFTER行类型(ROW_KIND=UPDATE_AFTER)的记录，表示更新后的数据状态

对于RabbitMQ和Console接收器，这两条记录都会被处理并输出，导致数据重复。而JDBC接收器在实现上会跳过UPDATE_BEFORE记录，只处理UPDATE_AFTER记录，因此不会出现重复。

技术分析

Postgres-CDC连接器基于Debezium实现变更数据捕获功能。当数据库表发生更新时，Debezium会捕获变更事件并生成两条记录：

• UPDATE_BEFORE: 表示更新前的数据状态
• UPDATE_AFTER: 表示更新后的数据状态

这种设计是为了完整记录数据变更历史，但在某些业务场景下，我们可能只关心变更后的最新状态。不同的接收器对这种变更事件的处理方式不同：

1. JDBC接收器：在实现上主动过滤掉了UPDATE_BEFORE记录，只处理UPDATE_AFTER记录
2. RabbitMQ和Console接收器：默认会处理所有类型的记录，包括UPDATE_BEFORE和UPDATE_AFTER

解决方案

方案一：使用FilterRowKind转换器

SeaTunnel提供了FilterRowKind转换器，可以过滤掉不需要的行类型。在配置文件中添加如下转换器配置：

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

然后将接收器的source_table_name指向过滤后的表名"employees_filtered"。

方案二：自定义接收器逻辑

如果使用的是自定义接收器，可以在接收器实现中检查SeaTunnelRow的ROW_KIND属性，只处理UPDATE_AFTER记录：

if (row.getRowKind() == SeaTunnelRowKind.UPDATE_AFTER) {
    // 处理记录
}

方案三：使用SQL转换过滤

如果配置中启用了SQL转换功能，可以使用SQL语句过滤掉不需要的行类型：

transform {
  Sql {
    source_table_name = "employees"
    result_table_name = "employees_filtered"
    query = "SELECT * FROM employees WHERE ROW_KIND <> 'UPDATE_BEFORE'"
  }
}

最佳实践建议

1. 根据业务需求决定是否需要保留变更前的数据。如果只需要最新状态，建议过滤掉UPDATE_BEFORE记录
2. 对于需要完整变更历史的场景，可以考虑将记录类型信息一并保存，便于后续分析
3. 在性能敏感场景下，尽早过滤不需要的记录可以减少网络传输和后续处理的开销
4. 不同接收器的一致性处理需要特别注意，建议在配置中明确记录过滤逻辑

总结

Postgres-CDC连接器产生的重复数据问题源于变更数据捕获机制本身的设计特点。通过合理使用SeaTunnel提供的转换器功能，可以灵活控制数据处理流程，满足不同业务场景的需求。理解CDC工作原理和SeaTunnel数据处理流程，有助于更好地设计和优化数据集成方案。