在NiFi中基于CSV指定列值新增行 拆分数组字段为多条独立行

以下两种方案都可以基于你已经去除括号的中间CSV文件实现需求： #

方案1：使用ExecuteScript处理器（Groovy脚本实现，逻辑简单高效） #

无需额外配置复杂的Schema，直接通过内置Groovy脚本处理，适配性更强：

• 添加ExecuteScript处理器，脚本引擎选择Groovy
• 脚本内容直接粘贴以下代码：

def flowFile = session.get()
if (!flowFile) return
try {
    flowFile = session.write(flowFile, { inputStream, outputStream ->
        def reader = new BufferedReader(new InputStreamReader(inputStream, "UTF-8"))
        def writer = new BufferedWriter(new OutputStreamWriter(outputStream, "UTF-8"))
        // 写入表头
        def header = reader.readLine()
        writer.writeLine(header)
        String line
        while ((line = reader.readLine()) != null) {
            // 按逗号分割字段并去除每个字段前后空格
            def parts = line.trim().split(",").collect { it.trim() }
            def id = parts[0]
            def name = parts[1]
            // 从第三个字段开始的所有值都属于Class列
            def classList = parts[2..-1]
            // 每个Class值单独生成一行
            classList.each { cls ->
                writer.writeLine("${id}, ${name}, ${cls}")
            }
        }
        writer.flush()
    } as StreamCallback)
    session.transfer(flowFile, REL_SUCCESS)
} catch (Exception e) {
    session.transfer(flowFile, REL_FAILURE)
}

方案2：使用标准处理器组合实现（无自定义脚本） #

如果不想使用脚本，可以按以下流程配置标准处理器：

1. ConvertRecord：将CSV转换为JSON格式
   • 配置CSVReader：Schema Access Strategy选择Use Schema Text，Schema文本填写：

     {
       "type": "record",
       "name": "Student",
       "fields": [
         {"name": "Id", "type": "string"},
         {"name": "Name", "type": "string"},
         {"name": "Class", "type": {"type": "array", "items": "string"}}
       ]
     }
     

   • 配置JsonRecordSetWriter：Schema策略选择继承自Reader的Schema即可
2. SplitJson：拆分Class数组生成单条记录
   • 配置JsonPath Expression为$.Class
3. JoltTransformJSON：对齐最终字段结构
   • Jolt转换规范填写：

     [
       {
         "operation": "shift",
         "spec": {
           "Id": "Id",
           "Name": "Name",
           "Class": "Class"
         }
       }
     ]
     

4. ConvertRecord：将JSON转换回CSV格式
   • 配置JsonTreeReader和CSVRecordSetWriter，Schema和第一步保持一致即可

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