NIFI ExecuteScript处理器Python过滤脚本无输出问题排查

核心错误分析 #

• 变量名笔误：代码中定义的过滤列表是filter，但判断条件里错误使用了未定义的patata变量，导致条件永远不成立，脚本执行时会抛出异常，后续逻辑无法正常运行，最终无FlowFile输出。
• 无效的FlowFile操作：在else分支中，先执行session.remove(flowFile)销毁了FlowFile，之后又调用session.transfer(flowFile, REL_FAILURE)，此时FlowFile对象已不存在，该操作不仅无效，还会引发错误导致脚本异常终止。

修正后的脚本 #

import json

flowFile = session.get()

if flowFile is not None:
    filter_list = ['1000007', '1000008']  # 避免使用Python内置关键字filter作为变量名
    
    try:
        # 读取并解析FlowFile的JSON内容
        content = flowFile.read().decode('utf-8')
        json_content = json.loads(content)
        
        # 检查Customer_AccountID是否在过滤列表内
        if json_content.get('Customer_AccountID') in filter_list:
            # 若无需修改内容，可跳过write步骤直接转移
            filtered_content = json.dumps(json_content)
            flowFile.write(bytearray(filtered_content, 'utf-8'))
            session.transfer(flowFile, REL_SUCCESS)
        else:
            # 直接转移到FAILURE分支，NiFi会自动处理FlowFile的移除
            session.transfer(flowFile, REL_FAILURE)
    except Exception as e:
        # 捕获JSON解析失败、字段缺失等异常，统一转移到FAILURE
        session.transfer(flowFile, REL_FAILURE)
        log.error(f"处理FlowFile出错: {str(e)}")

额外优化建议 #

• 避免使用Python内置关键字filter作为自定义变量名，改用filter_list这类名称，防止语法冲突。
• 添加异常捕获逻辑，处理JSON解析失败、目标字段不存在等场景，避免脚本因异常直接终止，同时便于排查问题。
• 如果不需要修改FlowFile内容，仅做过滤判断，可跳过write步骤，直接将符合条件的FlowFile转移到REL_SUCCESS，提升处理性能。

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