NiFi写入含数组JSON至MySQL表：如何保留关联ID？

问题场景 #

我刚开始用NiFi读取JSON数组并保存到MySQL表，核心需求是：JSON中的extraDetails是数组，需要将外层的id与每个extraDetails元素关联后插入表（表字段为id、from、to、additionalDetails）。

JSON示例：

[
  {
    "id": "2fa84997-b15f-4859-b315-8125ba79555f",
    "extraDetails": [
      {
        "from": "2023-10-06T03:05",
        "to": "2026-10-07T03:05:44",
        "additionalDetails": "{\"serviceId\":13,\"serviceTierIds\":[73,74],\"serviceEntitlementIds\":[]}"
      },
      {
        "from": "2023-10-06T03:06",
        "to": "2026-10-07T03:06:20",
        "additionalDetails": "{\"serviceId\":14,\"serviceTierIds\":[75,76,77,78],\"serviceEntitlementIds\":[]}"
      },
      {
        "from": "2023-10-06T03:06",
        "to": "2026-10-07T03:06:47",
        "additionalDetails": "{\"serviceId\":3,\"serviceTierIds\":[67,68],\"serviceEntitlementIds\":[]}"
      }
    ]
  },
  {
    "id": "cade90bd-62a3-48ce-87a0-0452e3efb3fc",
    "extraDetails": [
      {
        "from": "2023-10-27T11:00",
        "to": "2026-10-27T11:00:45",
        "additionalDetails": "{\"serviceId\":1,\"serviceTierIds\":[59,58,57,56,55,60,61],\"serviceEntitlementIds\":[]}"
      }
    ]
  }
]

之前尝试用SplitJson处理器，但会丢失外层的id属性，现在研究JoltTransformJSON处理器，想确认这是否是正确的解决方向？

解决方案：Jolt是正确方向 #

JoltTransformJSON可以一步完成外层ID与子数组元素的关联，比单独用SplitJson配合属性传递的方式更简洁高效。

1. 配置JoltTransformJSON的Shift规范 #

使用Shift类型的Jolt规则，将外层id复制到每个extraDetails元素中，输出平铺的对象数组：

[
  {
    "operation": "shift",
    "spec": {
      "*": {
        "extraDetails": {
          "*": {
            "@(2,id)": "[&1].id",
            "*": "[&1].&"
          }
        }
      }
    }
  }
]

执行后，JSON会被转换为符合插入要求的结构：

[
  {
    "id": "2fa84997-b15f-4859-b315-8125ba79555f",
    "from": "2023-10-06T03:05",
    "to": "2026-10-07T03:05:44",
    "additionalDetails": "{\"serviceId\":13,\"serviceTierIds\":[73,74],\"serviceEntitlementIds\":[]}"
  },
  {
    "id": "2fa84997-b15f-4859-b315-8125ba79555f",
    "from": "2023-10-06T03:06",
    "to": "2026-10-07T03:06:20",
    "additionalDetails": "{\"serviceId\":14,\"serviceTierIds\":[75,76,77,78],\"serviceEntitlementIds\":[]}"
  },
  {
    "id": "2fa84997-b15f-4859-b315-8125ba79555f",
    "from": "2023-10-06T03:06",
    "to": "2026-10-07T03:06:47",
    "additionalDetails": "{\"serviceId\":3,\"serviceTierIds\":[67,68],\"serviceEntitlementIds\":[]}"
  },
  {
    "id": "cade90bd-62a3-48ce-87a0-0452e3efb3fc",
    "from": "2023-10-27T11:00",
    "to": "2026-10-27T11:00:45",
    "additionalDetails": "{\"serviceId\":1,\"serviceTierIds\":[59,58,57,56,55,60,61],\"serviceEntitlementIds\":[]}"
  }
]

2. 后续插入MySQL的流程 #

• 若需要单条记录插入：接SplitJson处理器，设置Split Path为$.*，将每个对象拆分为单独的FlowFile；
• 结构化插入：使用ConvertRecord处理器，配置JsonTreeReader和AvroRecordSetWriter，再配合PutDatabaseRecord处理器，映射字段后直接批量插入MySQL；
• 自定义SQL插入：也可用PutSQL处理器，通过JsonPath提取字段构造INSERT语句，但批量场景下PutDatabaseRecord更高效。

为什么单独用SplitJson不行？ #

SplitJson仅拆分指定路径的数组，不会自动携带外层属性。若要单独用它，需先将外层id提取为FlowFile属性，再通过UpdateRecord或ReplaceText注入到每个子JSON中，步骤繁琐且易出错，远不如Jolt一步到位。

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