使用Debezium+Kafka同步PostgreSQL时如何获取操作类型及DDL数据？

一、如何获取操作类型（INSERT/UPDATE/DELETE） #

完全可以获取操作类型，Debezium输出的Kafka消息结构里自带操作标识，默认在 payload.op 字段：

• c：对应INSERT（CREATE）操作
• u：对应UPDATE操作
• d：对应DELETE操作
• r：对应初始快照的全量读取操作

你之前只拿到行数据，应该是只提取了消息里的 payload.after（更新后的数据）或 payload.before（更新前/删除前的数据）字段。可以在Jupyter里打印完整的消费消息，就能看到这个操作类型字段。示例代码：

# 打印完整消息结构以查看操作类型
for message in consumer:
    print(message.value())

二、如何捕获TRUNCATE、DROP、ALTER、DELETE等操作 #

1. DELETE操作捕获 #

DELETE属于DML操作，Debezium默认支持捕获，但需要确认以下配置：

• 确保PostgreSQL的 wal_level 设置为 logical（必须项）
• 连接器配置的 publication.autocreate.mode 包含DELETE，或手动创建的publication明确覆盖DELETE操作
• 消费时注意：DELETE事件的 payload.op 为 d，仅包含 payload.before 字段（存储被删除的行数据），无 payload.after

2. TRUNCATE、DROP、ALTER等操作捕获 #

• ALTER、DROP这类DDL操作：在连接器配置中添加 include.schema.changes=true，这类变更会被发送到专属的schema变更主题（默认命名格式为 {你的服务名}.schema-changes），消息内包含完整DDL语句与操作类型。
• TRUNCATE操作：需满足以下条件才能捕获：
  • PostgreSQL版本 ≥ 12
  • 连接器配置 plugin.name=pgoutput（使用PostgreSQL原生逻辑解码插件）
  • 添加配置 truncate.handling.mode=INCLUDE，TRUNCATE事件会被发送到对应数据主题，payload.op 为 t

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