Apache NiFi中Jolt转换嵌套数组的正确配置求助

输入数据 #

{
  "results": [
    {
      "group": {
        "queueId": "fila1",
        "mediaType": "message1"
      },
      "data": [
        {
          "metric": "oInteracting1",
          "stats": {
            "count": 1
          },
          "truncated": false
        },
        {
          "metric": "oWaiting1",
          "stats": {
            "count": 0
          }
        }
      ]
    },
    {
      "group": {
        "queueId": "fila2",
        "mediaType": "message2"
      },
      "data": [
        {
          "metric": "oInteracting2",
          "stats": {
            "count": 8
          },
          "truncated": false
        },
        {
          "metric": "oWaiting2",
          "stats": {
            "count": 0
          }
        }
      ]
    },
    {
      "group": {
        "queueId": "fila3",
        "mediaType": "message3"
      },
      "data": [
        {
          "metric": "oInteracting3",
          "stats": {
            "count": 1
          },
          "truncated": false
        },
        {
          "metric": "oWaiting3",
          "stats": {
            "count": 0
          }
        }
      ]
    }
  ]
}

失败的Jolt转换尝试 #

[
  {
    "operation": "shift",
    "spec": {
      "results": {
        "*": {
          "@group.queueId": "[&].queueid",
          "@group.mediaType": "[&].mediatype",
          "data": {
            "*": {
              "@metric": "[&3].metric1",
              "@stats.count": "[&3].metric1_count"
            }
          }
        }
      }
    }
  }
]

当前失败输出 #

[
  {
    "queueid": "fila1",
    "mediatype": "message1",
    "metric1": [
      "oInteracting1",
      "oWaiting1"
    ],
    "metric1_count": [
      1,
      0
    ]
  },
  {
    "queueid": "fila2",
    "mediatype": "message2",
    "metric1": [
      "oInteracting2",
      "oWaiting2"
    ],
    "metric1_count": [
      8,
      0
    ]
  },
  {
    "queueid": "fila3",
    "mediatype": "message3",
    "metric1": [
      "oInteracting3",
      "oWaiting3"
    ],
    "metric1_count": [
      1,
      0
    ]
  }
]

期望输出 #

[
  {
    "queueid": "fila1",
    "mediatype": "message1",
    "oInteracting1": 1,
    "oWaiting1": 0
  },
  {
    "queueid": "fila2",
    "mediatype": "message2",
    "oInteracting2": 8,
    "oWaiting2": 0
  },
  {
    "queueid": "fila3",
    "mediatype": "message3",
    "oInteracting3": 1,
    "oWaiting3": 0
  }
]

正确的Jolt转换配置 #

[
  {
    "operation": "shift",
    "spec": {
      "results": {
        "*": {
          "@group.queueId": "[&].queueid",
          "@group.mediaType": "[&].mediatype",
          "data": {
            "*": {
              "@stats.count": "[&3].@metric"
            }
          }
        }
      }
    }
  }
]

配置说明 #

• 采用shift操作完成字段重映射：
  • @group.queueId、@group.mediaType直接映射到对应队列对象的queueid、mediatype字段，[&]定位到当前results数组的索引位置。
  • 针对data数组内的每个指标项，通过@metric动态生成字段名，将@stats.count的值绑定到该动态字段；[&3]表示向上回溯3层（从data子元素到results数组项），确保指标字段精准对应到所属的队列对象中。

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