Kafka+Debezium同步PostgreSQL至QuestDB字段带前缀问题求助

你遇到的问题是因为Debezium生成的Kafka消息包含嵌套的before/after/source结构，QuestDB Sink Connector默认会将嵌套字段展开并添加前缀（如after_userid）。要直接提取after中的顶级字段，可通过以下两种方案解决：

方案一：在Sink Connector中配置unwrap转换（无需修改Source端） #

调整你的QuestDB Sink Connector配置，确保ExtractNewRecordState转换正确提取after中的内容作为消息的顶级结构：

{
  "name": "questdb-test-connector-latest-5",
  "config": {
    "connector.class": "io.questdb.kafka.QuestDBSinkConnector",
    "timestamp.kafka.native": "true",
    "client.conf.string": "http::addr=xyz:9000;",
    "topics": "xxx.xxx_questdb",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "include.key": "false",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "table": "test_table_trial_latest_7",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.delete.tombstone.handling.mode": "drop",
    // 可选：保留操作类型、时间戳等元数据，需添加以下参数
    "transforms.unwrap.add.fields": "op,ts_ms",
    "flatten.struct": "false"
  }
}

关键说明： #

• ExtractNewRecordState转换会自动剥离before/after/source外层结构，直接将after中的内容作为消息主体（适用于INSERT/UPDATE操作）
• delete.tombstone.handling.mode=drop会丢弃DELETE操作生成的墓碑消息，避免QuestDB写入空数据
• 若需要保留op（操作类型）、ts_ms（时间戳）等元数据，添加transforms.unwrap.add.fields参数即可，这些字段会成为QuestDB表的顶级字段

方案二：在Debezium Source Connector中提前处理（推荐） #

更高效的方式是在PostgreSQL的Debezium Source Connector中直接配置unwrap转换，让Kafka Topic中存储的就是after的原始数据，Sink端无需额外处理：

{
  "name": "postgresql-source-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "你的PG主机地址",
    "database.port": "5432",
    "database.user": "PG用户名",
    "database.password": "PG密码",
    "database.dbname": "kdb",
    "database.server.name": "test",
    "schema.include.list": "public",
    "table.include.list": "public.test_questdb",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.delete.tombstone.handling.mode": "drop"
  }
}

优势： #

• 减少Sink端的转换开销，Kafka消息结构更简洁
• 后续其他消费者使用该Topic时，无需再处理Debezium的嵌套结构

配置完成后，重启对应的Connector，同步到QuestDB的表字段就会直接是userid、firstname等原始字段，不再带有after_或source_前缀。

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