如何实现Kafka消费者将Debezium记录插入另一数据库并修改行数据？

方案一：基于Kafka Streams + JDBC Sink Connector（低代码快速落地） #

适合修改逻辑不复杂、希望快速搭建同步流程的场景：

• 解析并转换Debezium消息
  用Kafka Streams编写轻量拓扑，解析Debezium输出的JSON结构（提取after字段中的业务数据），完成字段修改（比如新增计算字段、调整字段值），然后将转换后的数据输出到新的Kafka主题。
  示例Java代码片段：

  StreamsBuilder builder = new StreamsBuilder();
  KStream<String, JsonNode> sourceStream = builder.stream("debezium-postgres-topic", Consumed.with(Serdes.String(), Serdes.Json()));
  KStream<String, JsonNode> transformedStream = sourceStream.mapValues(value -> {
      JsonNode after = value.get("after");
      // 示例：修改字段值、新增处理标记字段
      ObjectNode modified = (ObjectNode) after.deepCopy();
      modified.put("processed_timestamp", System.currentTimeMillis());
      modified.put("sync_status", "completed");
      return modified;
  });
  transformedStream.to("transformed-data-topic", Produced.with(Serdes.String(), Serdes.Json()));
  

• 配置JDBC Sink写入目标库
  使用Confluent JDBC Sink Connector（或Azure兼容连接器），订阅转换后的主题，配置目标数据库连接信息、表映射规则，开启upsert.mode以支持插入/更新操作，确保主键匹配目标表主键字段。

方案二：自定义Kafka消费者（灵活可控） #

适合修改逻辑复杂、需要完全自定义业务流程的场景：

• 选择开发语言与客户端
  基于熟悉的语言（Java/Python/Go等），使用官方Kafka客户端（如Java kafka-clients、Python confluent-kafka）订阅Debezium主题。
• 解析Debezium消息结构
  每条消息包含before（变更前数据）、after（变更后数据）、op（操作类型：c=插入、u=更新、d=删除）等字段，根据业务需求提取after数据，按需处理或忽略d类型的删除操作。
• 数据修改与写入
  对after中的数据做自定义修改（如字段格式转换、业务规则计算），然后通过JDBC驱动或ORM框架（MyBatis/SQLAlchemy）将数据插入/更新到目标数据库。
• 关键保障措施
  • 开启手动offset提交：确保数据成功写入数据库后再提交消费偏移量，避免消息丢失。
  • 处理幂等性：利用目标表主键去重，或开启数据库幂等写入机制，防止重复消费导致数据重复。
  • 事务支持：若业务要求严格一致性，将消费消息、修改数据、写入数据库封装在一个事务中。

方案三：基于Kafka Connect SMT（零代码极简修改） #

仅适合字段重命名、固定值添加、简单值转换等极简修改场景：

• 直接在JDBC Sink Connector中配置Single Message Transforms（SMT），无需编写代码即可完成数据修改：
  示例配置（字段重命名+新增固定标记）：

  transforms=renameField,addProcessedFlag
  transforms.renameField.type=org.apache.kafka.connect.transforms.ReplaceField$Value
  transforms.renameField.renames=old_user_id:user_id
  transforms.addProcessedFlag.type=org.apache.kafka.connect.transforms.InsertField$Value
  transforms.addProcessedFlag.static.field=is_synced
  transforms.addProcessedFlag.static.value=true
  

• 注意：SMT需要针对Debezium的消息结构指定字段路径（如after.old_user_id），确保修改作用在正确的业务数据上。

通用注意事项 #

• 错误处理：配置死信队列（DLQ），将处理失败的消息转发至单独主题，便于后续排查与重试。
• 性能优化：采用批量消费+批量写入数据库的方式，减少IO开销；调整消费者并发数，提升处理吞吐量。
• 格式兼容：Debezium默认输出JSON格式，若使用Avro需配置Schema Registry，确保消费者能正确解析消息。

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