如何在Apache NiFi中用JOLT将嵌套JSON展平为单值对象？

需要在Apache NiFi中使用JOLTTransformation处理器处理嵌套JSON，将每个dataseries元素与外层的product、init字段组合成独立对象，以便后续通过ConvertJsonToSql处理器插入PostgreSQL数据库。当前使用的JOLT Spec生成了聚合数组形式的输出，不符合需求。

输入JSON #

{
  "product": "astro",
  "init": "2022091400",
  "dataseries": [
    {
      "timepoint": 3,
      "cloudcover": 2,
      "seeing": 6,
      "transparency": 2,
      "lifted_index": 2,
      "rh2m": 3,
      "wind10m": {
        "direction": "N",
        "speed": 3
      },
      "temp2m": 33,
      "prec_type": "none"
    },
    {
      "timepoint": 6,
      "cloudcover": 2,
      "seeing": 6,
      "transparency": 2,
      "lifted_index": 2,
      "rh2m": 1,
      "wind10m": {
        "direction": "NW",
        "speed": 3
      },
      "temp2m": 35,
      "prec_type": "none"
    },
    {
      "timepoint": 9,
      "cloudcover": 1,
      "seeing": 6,
      "transparency": 2,
      "lifted_index": 2,
      "rh2m": 2,
      "wind10m": {
        "direction": "N",
        "speed": 3
      },
      "temp2m": 35,
      "prec_type": "none"
    }
  ]
}

当前JOLT Spec #

[
  {
    "operation": "shift",
    "spec": {
      "product": "product",
      "init": "init",
      "dataseries": {
        "*": {
          "timepoint": "timepoint",
          "cloudcover": "cloudcover",
          "seeing": "seeing",
          "transparency": "transparency",
          "lifted_index": "lifted_index",
          "rh2m": "rh2m",
          "wind10m": {
            "direction": "direction",
            "speed": "speed"
          },
          "temp2m": "temp2m",
          "prec_type": "prec_type"
        }
      }
    }
  }
]

实际输出 #

{
  "product" : "astro",
  "init" : "2022091400",
  "timepoint" : [ 3, 6, 9 ],
  "cloudcover" : [ 2, 2, 1 ],
  "seeing" : [ 6, 6, 6 ],
  "transparency" : [ 2, 2, 2 ],
  "lifted_index" : [ 2, 2, 2 ],
  "rh2m" : [ 3, 1, 2 ],
  "direction" : [ "N", "NW", "N" ],
  "speed" : [ 3, 3, 3 ],
  "temp2m" : [ 33, 35, 35 ],
  "prec_type" : [ "none", "none", "none" ]
}

预期输出 #

[
  {
    "product" : "astro",
    "init" : "2022091400",
    "timepoint" : 3,
    "cloudcover" : 2,
    "seeing" : 6,
    "transparency" : 2,
    "lifted_index" : 2,
    "rh2m" : 3,
    "direction" : "N",
    "speed" : 3,
    "temp2m" : 33,
    "prec_type" : "none"
  },
  {
    "product" : "astro",
    "init" : "2022091400",
    "timepoint" : 6,
    "cloudcover" : 2,
    "seeing" : 6,
    "transparency" : 2,
    "lifted_index" : 2,
    "rh2m" : 1,
    "direction" : "NW",
    "speed" : 3,
    "temp2m" : 35,
    "prec_type" : "none"
  },
  {
    "product" : "astro",
    "init" : "2022091400",
    "timepoint" : 9,
    "cloudcover" : 1,
    "seeing" : 6,
    "transparency" : 2,
    "lifted_index" : 2,
    "rh2m" : 2,
    "direction" : "N",
    "speed" : 3,
    "temp2m" : 35,
    "prec_type" : "none"
  }
]

修正后的JOLT Spec #

[
  {
    "operation": "shift",
    "spec": {
      "dataseries": {
        "*": {
          "@(2,product)": "[&1].product",
          "@(2,init)": "[&1].init",
          "timepoint": "[&1].timepoint",
          "cloudcover": "[&1].cloudcover",
          "seeing": "[&1].seeing",
          "transparency": "[&1].transparency",
          "lifted_index": "[&1].lifted_index",
          "rh2m": "[&1].rh2m",
          "wind10m": {
            "direction": "[&2].direction",
            "speed": "[&2].speed"
          },
          "temp2m": "[&1].temp2m",
          "prec_type": "[&1].prec_type"
        }
      }
    }
  }
]

说明 #

• 使用@(2,product)和@(2,init)向上两级引用外层的product和init字段值；
• [&1]表示以当前dataseries元素的索引作为输出数组的位置，将外层字段与当前元素的所有字段合并，生成独立的对象；
• 处理后的输出为对象数组，可直接被ConvertJsonToSql处理器识别，将每个对象作为一条记录插入PostgreSQL数据库。

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