Apache Kafka ServiceNow源连接器JSON反扁平化配置求助

Kafka Connect没有内置的"反扁平化"转换，但可以通过Single Message Transform (SMT) 的组合或自定义脚本实现从ServiceNow扁平字段到嵌套JSON的转换，以下是两种可行方案：

方案1：内置SMT组合实现固定结构嵌套 #

适用于结构简单的嵌套需求，比如将u_payload_projectid放入payload.projectId层级，u_messagetype作为顶级字段：

在ServiceNow源连接器的配置中添加以下transforms段：

transforms=unwrap, nestPayload, renameFields
# 提取Debezium包装后的实际记录（如果使用Debezium ServiceNow连接器）
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
# 将指定字段嵌套到payload层级
transforms.nestPayload.type=org.apache.kafka.connect.transforms.HoistField$Value
transforms.nestPayload.field=payload
# 重命名扁平字段为嵌套结构中的目标字段名
transforms.renameFields.type=org.apache.kafka.connect.transforms.ReplaceField$Value
transforms.renameFields.renames=u_payload_projectid:projectId,u_messagetype:messageType

如果仅需部分字段嵌套，可先通过Filter SMT保留目标字段，再执行嵌套操作。

方案2：ScriptTransform自定义复杂嵌套逻辑 #

如果需要多层嵌套或灵活的字段映射，用自定义脚本完全控制结构：

配置示例：

transforms=customNest
transforms.customNest.type=org.apache.kafka.connect.transforms.ScriptTransform$Value
transforms.customNest.script.type=js
transforms.customNest.script=
  var result = {};
  // 定义顶级字段
  result.messageType = value.u_messagetype;
  result.sys_id = value.sys_id;
  // 构建嵌套的payload结构
  result.payload = {
    projectId: value.u_payload_projectid,
    // 按需添加更多嵌套字段，比如u_payload_status: status
  };
  return result;

脚本中value代表原始记录的Value部分，可直接访问ServiceNow表的扁平字段，自由组装嵌套结构。

验证与调试 #

用Kafka控制台消费者确认输出结构是否符合预期：

kafka-console-consumer.sh --bootstrap-server <kafka-broker>:9092 --topic <your-target-topic> --from-beginning --property print.key=true --property value.deserializer=org.apache.kafka.connect.json.JsonConverter --property value.converter.schemas.enable=false

若使用Debezium连接器，务必先通过ExtractNewRecordState提取实际数据，避免处理Debezium的包装结构。

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