如何在Kafka Connector中从响应体提取指定ID字段

你可以通过在Kafka Connect的连接器配置中添加额外的Transforms来实现只保留Id字段、移除其他内容的需求，具体有两种常用方式：

方式一：使用ExtractField Transform #

这个Transform可以直接从记录中提取指定的单个字段，将整个记录替换为该字段的值（最终输出纯Id数值，比如20030726）。修改你的连接器配置如下：

{
  "name": "****",
  "config": {
    // 原有配置保持不变
    "transforms": "Reroute, unwrap, ExtractId",
    "transforms.ExtractId.type": "org.apache.kafka.connect.transforms.ExtractField$Value",
    "transforms.ExtractId.field": "Id"
  }
}

方式二：使用ReplaceField Transform #

如果你希望最终输出是仅包含Id字段的JSON对象（比如{"Id":20030726}），可以用这个Transform的白名单功能过滤字段：

{
  "name": "****",
  "config": {
    // 原有配置保持不变
    "transforms": "Reroute, unwrap, KeepOnlyId",
    "transforms.KeepOnlyId.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.KeepOnlyId.whitelist": "Id"
  }
}

关键注意点 #

• 保持transform执行顺序正确：必须先完成Reroute路由和unwrap解包操作，再执行字段提取/过滤的transform，否则会无法正确识别目标字段。
• 配置修改后，重启连接器或重新提交配置即可生效，可通过Kafka消费者查看输出结果验证效果。

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