NiFi：无需脚本实现拆分属性字符串并按序填充JSON数组

需求描述 #

现有一个JSON数组，每个对象包含id和status字段；同时存在一个名为task-ids的属性，值为逗号分隔的数字字符串，其元素数量和顺序与JSON数组中的对象完全对应。需要将该属性的元素按顺序添加为JSON数组中每个对象的taskId字段。

输入JSON数组示例 #

[
  {
    "id": 1,
    "status": 0
  },
  {
    "id": 2,
    "status": 0
  },
  {
    "id": 3,
    "status": 1
  }
]

task-ids属性说明 #

属性值为逗号分隔的数字字符串，例如500, 501, 502、765, 780, 790等，元素数量和顺序与JSON数组的对象一一对应。

预期输出 #

[
  {
    "id": 1,
    "status": 0,
    "taskId": 500
  },
  {
    "id": 2,
    "status": 0,
    "taskId": 501
  },
  {
    "id": 3,
    "status": 1,
    "taskId": 502
  }
]

现有Groovy脚本实现 #

def numbersString = flowFile.getAttribute("task-ids");
def numbersArray = numbersString.split(',').collect { it as Integer }

for (int i = 0; i < numbersArray.size(); i++) {
    jsonArray[i].taskId = numbersArray[i]
}

无脚本替代实现方案（以Apache NiFi为例） #

通过组合以下处理器可实现需求，全程无需编写自定义脚本：

• SplitJson：将输入的JSON数组拆分为单个JSON对象流文件。配置JsonPath Expression为$[*]，每个数组元素会转为独立的flowFile，同时自动添加fragment.index属性记录原数组中的位置索引（从0开始）。
• UpdateAttribute：处理task-ids属性，添加自定义属性：
  • 名称：task-id-list，值：${task-ids:replaceAll('\\s+', ''):split(',')}
    该配置会去除字符串中的空格并将其拆分为可索引的列表。
• UpdateRecord：为单个JSON对象添加taskId字段。使用JsonTreeReader和JsonRecordSetWriter作为读写器，新增字段taskId，值设置为${task-id-list:get(${fragment.index})}，通过fragment.index获取对应位置的task-id值。
• MergeRecord：将处理后的单个JSON对象重新合并为数组。配置Record Reader为JsonTreeReader，Record Writer为JsonRecordSetWriter，依赖fragment.index保证合并顺序与原数组一致。

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