如何通过NiFi JOLTTransformRecord将扁平JSON转为嵌套树形结构？

问题场景 #

使用NiFi的JOLTTransformRecord处理器，将扁平JSON数组转换为包含嵌套日志数组的树形JSON结构时遇到转换异常，以下是相关数据及当前配置，需修正JOLT转换规则。

源JSON #

[
  {
    "log_date": "2024-07-24 16:10:22.851",
    "log_level": "INFO",
    "log_message": "Message1",
    "request_id": "0a3d546d"
  },
  {
    "log_date": "2024-07-24 16:10:22.851",
    "log_level": "INFO",
    "log_message": "Message2",
    "request_id": "0a3d546d"
  },
  {
    "log_date": "2024-07-24 16:10:22.851",
    "log_level": "INFO",
    "log_message": "Message3",
    "request_id": "0a3d546d"
  },
  {
    "log_date": "2024-07-24 16:10:22.572",
    "log_level": "INFO",
    "log_message": "Message4",
    "request_id": "0a3d546d"
  },
  {
    "log_date": "2024-07-24 16:10:22.572",
    "log_level": "INFO",
    "log_message": "Message5",
    "request_id": "0a3d546d"
  },
  {
    "log_date": "2024-07-24 16:10:22.572",
    "log_level": "INFO",
    "log_message": "Message6",
    "request_id": "0a3d546d"
  }
]

期望输出 #

{
  "request_id": "0a3d546d",
  "log": [
    {
      "log_date": "2024-07-24 16:10:22.851",
      "log_level": "INFO",
      "log_message": "Message1"
    },
    {
      "log_date": "2024-07-24 16:10:22.851",
      "log_level": "INFO",
      "log_message": "Message2"
    },
    {
      "log_date": "2024-07-24 16:10:22.851",
      "log_level": "INFO",
      "log_message": "Message3"
    },
    {
      "log_date": "2024-07-24 16:10:22.572",
      "log_level": "INFO",
      "log_message": "Message4"
    },
    {
      "log_date": "2024-07-24 16:10:22.572",
      "log_level": "INFO",
      "log_message": "Message5"
    },
    {
      "log_date": "2024-07-24 16:10:22.572",
      "log_level": "INFO",
      "log_message": "Message6"
    }
  ]
}

当前JOLT规范问题分析 #

1. Shift操作逻辑错误："log*": { "@(1)": "log" }会匹配所有以log开头的字段，重复将整个源对象写入log数组，导致数据冗余。
2. Cardinality配置错误：第二个Cardinality操作中目标字段logs拼写错误（应为log），且字段组合写法不符合JOLT语法规范。

正确JOLT转换规范 #

[
  {
    "operation": "shift",
    "spec": {
      "*": {
        "request_id": "request_id",
        "log_date": "log[&1].log_date",
        "log_level": "log[&1].log_level",
        "log_message": "log[&1].log_message"
      }
    }
  },
  {
    "operation": "cardinality",
    "spec": {
      "request_id": "ONE"
    }
  }
]

规则说明

• Shift操作：遍历源数组每个元素（*），将request_id统一映射到顶层字段；将每个元素的log_date、log_level、log_message分别映射到log数组的对应索引位置（&1代表当前元素在源数组中的索引）。
• Cardinality操作：将重复赋值的request_id从数组转换为单个值（因所有源元素的request_id相同，ONE规则会保留唯一有效值）。

内容的提问来源于stack exchange，提问作者Будён Михайлович Семённый