NiFi中用ExecuteScript提取整数数组最大值报错求助

问题背景 #

上游EvaluateJsonPath处理器输出整数数组[9308023, 48243, 429402, 589348934, 4943, 4298040240, 424820482]，需要用下游ExecuteScript处理器的Python代码提取数组最大值并传递给下一个处理器，但运行代码后出现报错。

报错信息 #

ExecuteScript[id=018810db-110d-17c5-17da-0fdf1ca49296] 处理流文件失败：read()：第二个参数无法转换为org.apache.nifi.processor.io.InputStreamCallback

错误原因 #

1. 回调类型不匹配：session.read()方法要求传入InputStreamCallback类型的回调，但代码中传入的是StreamCallback（同时处理输入输出的回调类），导致类型强制转换失败。
2. 逻辑不符合需求：原代码对数组排序后取第一个元素，实际得到的是最小值，并非需求的最大值。
3. 冗余调用：同时调用session.read()和session.write()，StreamCallback只需通过session.write()即可完成输入读取和输出写入操作。

修正后的代码 #

from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets
from org.apache.nifi.processor.io import StreamCallback

class PyStreamCallback(StreamCallback):
    def process(self, inputStream, outputStream):
        # 读取流文件内容并去除首尾的方括号
        flowFileText = IOUtils.toString(inputStream, StandardCharsets.UTF_8).strip().strip('[]')
        
        if flowFileText:
            # 分割字符串并转换为整数数组
            ids = [int(id.strip()) for id in flowFileText.split(',')]
            # 直接获取数组最大值
            max_id = max(ids)
            # 将最大值写入输出流
            outputStream.write(str(max_id))
        else:
            outputStream.write("")

streamCallback = PyStreamCallback()

flowFile = session.get()
if flowFile is not None:
    try:
        # 仅通过session.write()调用StreamCallback，同时完成读和写操作
        session.write(flowFile, streamCallback)
        session.transfer(flowFile, REL_SUCCESS)
    except Exception as e:
        log.error("处理流文件失败: " + str(e))
        session.transfer(flowFile, REL_FAILURE)
    finally:
        session.commit()

关键修改点 #

• 移除session.read()调用，仅使用session.write()配合StreamCallback完成输入读取与输出写入，解决类型不匹配问题。
• 使用max()函数直接获取数组最大值，替代排序取首元素的低效逻辑，精准满足需求。
• 增加去除数组首尾方括号的处理，避免分割时出现无效字符导致整数转换失败。

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