Apache NiFi中Groovy脚本出现意外字符'('错误求助

问题描述 #

在Apache NiFi中编写Groovy脚本，意图将jsonData属性中的"All"值替换为response属性对应的数据，但执行时第22行出现「意外字符'('」编译错误，请求排查脚本问题。

原脚本代码 #

import groovy.json.JsonBuilder
import groovy.json.JsonSlurper
import org.apache.commons.io.IOUtils
import org.apache.groovy.json.internal.LazyMap
import org.apache.commons.io.IOUtils
import org.apache.nifi.processor.io.StreamCallback
import java.nio.charset.StandardCharsets

def flowfile=session.get();
if(flowfile==null){return;}
flowfile=session.write(flowfile,{inputStream,outputStream->
def content=IOUtils.toString(inputStream,StandardCharsets.UTF_8)
def inJson=new JsonSlurper().parseText(content)
String apiResult=flowfile.getAttribute("response")
String S3data1=flowfile.getAttribute("jsonData")
def apiResponse=new JsonSlurper().parseText(apiResult)
def S3data=new JsonSlurper().parseText(S3data1)
def result=apiResponse.data.collect{i->S3data.collectEntries{k,v->=="All"?i[0]:v}}
outputStream.write(result.toString().getBytes(StandardCharsets.UTF_8))}as StreamCallback)
session.transfer(flowfile,REL_SUCCESS)

错误日志 #

ERROR [Timer-Driven Process Thread-9] o.a.nifi.processors.script.ExecuteScript ExecuteScript[id=75571200-41b6-3e1e-1a48-c2abb37207f1] Processing failed
org.apache.nifi.processor.exception.ProcessException: javax.script.ScriptException: org.codehaus.groovy.control.MultipleCompilationErrorsException: startup failed:
Script5192.groovy: 22: Unexpected input: '(' @ line 22, column 23.
   flowfile=session.write(flowfile,{inputStream,outputStream->
                         ^

1 error
...

response属性数据 #

{"status":"success","data":[[446136,0],[446151,0],...[237082,1]]}

排查与修复方案 #

1. 修复第22行语法歧义 #

原脚本中session.write的闭包参数写法存在解析歧义，Groovy无法正确识别参数结构。调整闭包的声明格式，在闭包前添加空格，并确保as StreamCallback的位置符合语法规范。

2. 删除重复导入 #

原脚本重复导入了org.apache.commons.io.IOUtils，删除其中一个重复导入项。

3. 修复collectEntries逻辑语法错误 #

原代码中==="All"?i[0]:v缺少比较对象，应明确为v == "All"；同时collectEntries需要返回键值对结构，补充[(k): ...]的Map条目格式。

修复后的完整脚本 #

import groovy.json.JsonBuilder
import groovy.json.JsonSlurper
import org.apache.commons.io.IOUtils
import org.apache.groovy.json.internal.LazyMap
import org.apache.nifi.processor.io.StreamCallback
import java.nio.charset.StandardCharsets

def flowfile = session.get()
if (!flowfile) {
    return
}

flowfile = session.write(flowfile, { inputStream, outputStream ->
    def content = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
    def inJson = new JsonSlurper().parseText(content)
    String apiResult = flowfile.getAttribute("response")
    String S3data1 = flowfile.getAttribute("jsonData")
    def apiResponse = new JsonSlurper().parseText(apiResult)
    def S3data = new JsonSlurper().parseText(S3data1)
    
    def result = apiResponse.data.collect { i ->
        S3data.collectEntries { k, v ->
            [(k): v == "All" ? i[0] : v]
        }
    }
    
    outputStream.write(result.toString().getBytes(StandardCharsets.UTF_8))
} as StreamCallback)

session.transfer(flowfile, REL_SUCCESS)

额外优化说明 #

• 调整代码格式，增加换行与缩进，提升可读性
• 将if(flowfile==null)简化为if (!flowfile)，符合Groovy简洁语法规范

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