如何在Apache NiFi中用Jolt将JSON数组转为嵌套结构？

输入JSON #

[
  {
    "LocationAvailabilitySync": {
      "TransactionNumber": "31ea1a10-abc4-4aa5-a8c6-6f204903073a",
      "TransactionType": "SyncDetail",
      "CurrentPage": 5,
      "CurrentCount": 15,
      "TotalCount": 9,
      "TotalPages": 9,
      "SyncCount": null,
      "ViewConfigurationName": null
    },
    "ItemId": "5678",
    "LocationId": "0088",
    "Status": "AVAILABLE",
    "ViewName": "ATC_1",
    "ViewId": "05fdb789-a372-4f0e-8e35-9517f3cf02cc",
    "Quantity": 567,
    "TransactionDateTime": "2020-04-29T18:35:56.662",
    "NextAvailabilityDate": null,
    "TotalIncludingSubstituteItems": 567,
    "SubstituteItemsAvailable": false,
    "SubstitutionDetail": null
  },
  {
    "LocationAvailabilitySync": {
      "TransactionNumber": "31ea1a10-abc4-4aa5-a8c6-6f204903073a",
      "TransactionType": "SyncDetail",
      "CurrentPage": 5,
      "CurrentCount": 15,
      "TotalCount": 9,
      "TotalPages": 9,
      "SyncCount": null,
      "ViewConfigurationName": null
    },
    "ItemId": "1234",
    "LocationId": "0088",
    "Status": "AVAILABLE",
    "ViewName": "ATC_1",
    "ViewId": "05fdb789-a372-4f0e-8e35-9517f3cf02cc",
    "Quantity": 567,
    "TransactionDateTime": "2020-04-29T18:35:56.662",
    "NextAvailabilityDate": null,
    "TotalIncludingSubstituteItems": 567,
    "SubstituteItemsAvailable": false,
    "SubstitutionDetail": null
  }
]

期望输出JSON #

{
  "LocationAvailabilitySync": [
    {
      "TransactionNumber": "31ea1a10-abc4-4aa5-a8c6-6f204903073a",
      "TransactionType": "SyncDetail",
      "CurrentPage": 5,
      "CurrentCount": 15,
      "TotalCount": 9,
      "TotalPages": 9,
      "SyncCount": null,
      "ViewConfigurationName": null,
      "LocationAvailabilityResponse": [
        {
          "ItemId": "5678",
          "LocationId": "0088",
          "Status": "AVAILABLE",
          "ViewName": "ATC_SOFKA1",
          "ViewId": "e88130a2-6a07-408b-8bf8-03e3d28cf057",
          "Quantity": 567,
          "TransactionDateTime": "2020-04-29T18:35:56.662",
          "NextAvailabilityDate": null,
          "TotalIncludingSubstituteItems": 567,
          "SubstituteItemsAvailable": false,
          "SubstitutionDetail": null
        },
        {
          "ItemId": "1234",
          "LocationId": "0088",
          "Status": "AVAILABLE",
          "ViewName": "ATC_SOFKA2",
          "ViewId": "e88130a2-6a07-408b-8bf8-03e3d28cf058",
          "Quantity": 567,
          "TransactionDateTime": "2020-04-29T18:35:56.662",
          "NextAvailabilityDate": null,
          "TotalIncludingSubstituteItems": 567,
          "SubstituteItemsAvailable": false,
          "SubstitutionDetail": null
        }
      ]
    }
  ]
}

Jolt转换规范 #

[
  // 第一步：重组结构，提取公共字段并收集子项
  {
    "operation": "shift",
    "spec": {
      "*": {
        "LocationAvailabilitySync": {
          "*": "LocationAvailabilitySync[0].&"
        },
        "*": "LocationAvailabilitySync[0].LocationAvailabilityResponse[&1].&"
      }
    }
  },
  // 第二步：修改ViewName和ViewId为目标格式
  {
    "operation": "modify-overwrite-beta",
    "spec": {
      "LocationAvailabilitySync": {
        "*": {
          "LocationAvailabilityResponse": {
            "*": {
              "ViewName": "=concat('ATC_SOFKA', @(1) + 1)",
              "ViewId": "=concat('e88130a2-6a07-408b-8bf8-03e3d28cf05', @(1) + 7)"
            }
          }
        }
      }
    }
  }
]

转换说明 #

• Shift操作：
  • 遍历输入数组的每个元素，将所有元素共用的LocationAvailabilitySync字段提取到输出的LocationAvailabilitySync数组第一个元素中（因输入中该字段内容完全一致，最终仅保留一份）。
  • 将每个元素中除LocationAvailabilitySync外的所有字段，按原数组索引位置收集到LocationAvailabilityResponse数组中。
• Modify操作：
  • 针对LocationAvailabilityResponse中的每个元素，根据其数组索引（从0开始）生成目标格式的ViewName和ViewId。若实际场景中这两个字段的生成规则不同，可直接修改concat函数内的表达式。

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