Apache NiFi中使用Jolt Processor转换复杂JSON数据的需求

输入JSON #

{
  "subscriptionId": "63",
  "data": [
    {
      "type": "demo",
      "Data": {
        "transactionId": "598958",
        "type": "json",
        "xyz": "pqr",
        "name": "john"
      },
      "nameData": [
        {
          "name": "Ama",
          "nameCount": "215",
          "genderData": [
            {
              "gender": "Male",
              "genderCount": "140"
            },
            {
              "gender": "Female",
              "genderCount": "75"
            }
          ]
        },
        {
          "name": "Aedes",
          "nameCount": "161",
          "genderData": [
            {
              "gender": "Female",
              "genderCount": "134"
            },
            {
              "gender": "Male",
              "genderCount": "27"
            }
          ]
        },
        {
          "name": "Culex",
          "nameCount": "2610",
          "genderData": [
            {
              "gender": "Male",
              "genderCount": "1926"
            },
            {
              "gender": "Female",
              "genderCount": "684"
            }
          ]
        },
        {
          "name": "Kamp",
          "nameCount": "465",
          "genderData": [
            {
              "gender": "Male",
              "genderCount": "465"
            }
          ]
        }
      ]
    }
  ]
}

期望输出JSON #

{
    "transactionId": "598958",
    "type": "json",
    "xyz": "pqr",
    "name": "john",
    "alert_array": [{
        "abc": "123",
        "xyz": "pqrs",
        "properties": [{
                "key": "Ama",
                "value": "215",
                "object": [{
                        "key": "Male",
                        "value": "140"
                    },
                    {
                        "key": "Female",
                        "value": "75"
                    }
                ]
            },
            {
                "key": "Aedes",
                "value": "161",
                "object": [{
                        "key": "Male",
                        "value": "134"
                    },
                    {
                        "key": "Female",
                        "value": "27"
                    }
                ]
            },
            {
                "key": "Culex",
                "value": "2610",
                "object": [{
                        "key": "Male",
                        "value": "1926"
                    },
                    {
                        "key": "Female",
                        "value": "684"
                    }
                ]
            },
            {
                "key": "Kamp",
                "value": "465",
                "object": [{
                    "key": "Male",
                    "value": "465"
                }]
            },
            {
                "key": "type",
                "value": "demo"
            }
        ]
    }]
}

Jolt 转换规则 #

以下是实现该转换的Jolt Spec：

[
  {
    "operation": "shift",
    "spec": {
      "data": {
        "*": {
          "Data": {
            "*": "&"
          },
          "nameData": {
            "*": {
              "name": "alert_array[0].properties[&1].key",
              "nameCount": "alert_array[0].properties[&1].value",
              "genderData": {
                "*": {
                  "gender": "alert_array[0].properties[&3].object[&1].key",
                  "genderCount": "alert_array[0].properties[&3].object[&1].value"
                }
              }
            }
          },
          "type": "alert_array[0].properties[-1].key",
          "#demo": "alert_array[0].properties[-1].value"
        }
      }
    }
  },
  {
    "operation": "default",
    "spec": {
      "alert_array": [
        {
          "abc": "123",
          "xyz": "pqrs"
        }
      ]
    }
  }
]

规则说明 #

1. Shift 操作：负责核心的数据提取与重组
   • 将data[0].Data下的所有字段直接映射到根节点，对应输出的transactionId、type等字段；
   • 遍历nameData数组，把每个元素的name、nameCount映射到alert_array[0].properties的对应位置；
   • 嵌套遍历genderData数组，将gender和genderCount映射到对应properties元素的object数组中；
   • 把data[0].type的值添加到properties数组的最后一个元素。
2. Default 操作：为alert_array添加固定的abc和xyz字段值。

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