如何在NiFi中按需拆分Flow File的JSON数组

针对你遇到的问题——仅在JSON数组包含多个对象时拆分，单个对象/单元素数组保留原结构，提供两种可行方案：

方案一：多处理器分支路由（直观易维护） #

1. 识别JSON结构与数组长度 #

添加EvaluateJsonPath处理器，配置：

• 目标(Destination)：flowfile-attribute
• 新增属性：
  • is_array：表达式 $ instanceof Array，用于判断输入是否为JSON数组
  • array_size：表达式 $.length()，仅当输入为数组时会生成该属性，值为数组元素数量

2. 按规则路由FlowFile #

添加RouteOnAttribute处理器，配置两个路由规则：

• split_needed：匹配条件 ${is_array:equals('true')} and ${array_size:gt(1)}
• keep_original：匹配条件 ${is_array:equals('false')} or ${array_size:le(1)}

3. 分支处理 #

• 将split_needed路由连接到SplitJson处理器，设置JsonPath表达式为 $.*，实现多元素数组拆分
• 将keep_original路由直接指向成功关系，输出原FlowFile（单个对象或单元素数组均保留原结构）

方案二：ExecuteScript脚本一次性处理（简洁高效） #

添加ExecuteScript处理器，选择Groovy作为脚本语言，粘贴以下代码：

import groovy.json.JsonSlurper
import groovy.json.JsonBuilder

def flowFile = session.get()
if (!flowFile) return

def slurper = new JsonSlurper()
def jsonContent = slurper.parseText(flowFile.read().getText())

if (jsonContent instanceof List && jsonContent.size() > 1) {
    // 多元素数组，拆分每个对象为独立FlowFile
    jsonContent.each { item ->
        def outFile = session.create(flowFile)
        outFile.write("${new JsonBuilder(item).toPrettyString()}")
        session.transfer(outFile, REL_SUCCESS)
    }
    session.remove(flowFile)
} else {
    // 单个对象或单元素数组，直接输出原内容
    session.transfer(flowFile, REL_SUCCESS)
}

该脚本会自动判断输入JSON的类型：

• 若为多元素数组，拆分每个对象为独立FlowFile
• 若为单个JSON对象或单元素数组，直接保留原结构输出

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