在NiFi中使用Python3访问FlowFile属性与内容的技术问询

核心限制说明 #

NiFi的ExecuteScript组件原生仅支持Jython 2.7，无法直接在该组件内运行Python3代码并访问NiFi的session对象或FlowFile属性。若要使用Python3，只能通过ExecuteStreamCommand组件调用外部Python3脚本，但这种方式无法直接访问NiFi的session对象，仅能通过标准输入输出处理FlowFile内容，通过环境变量传递FlowFile属性。

依赖库与具体实现 #

1. 读取与输出FlowFile内容 #

无需额外依赖库，直接使用Python标准库sys即可，示例代码：

import sys

# 从标准输入读取FlowFile内容
for line in sys.stdin:
    # 这里添加自定义处理逻辑，示例：去除首尾空白
    processed_content = line.strip()
    # 将处理后的内容输出到标准输出，作为新FlowFile的内容
    print(processed_content)

2. 获取FlowFile属性 #

NiFi会将FlowFile的所有属性转换为NIFI_ATTRIBUTE_<属性名>格式的环境变量（属性名转为大写，特殊字符替换为下划线），可通过Python标准库os读取：

import os
import sys

# 读取FlowFile的filename属性
target_filename = os.getenv("NIFI_ATTRIBUTE_FILENAME")
if target_filename:
    # 输出日志到标准错误流，NiFi会捕获并展示在组件日志中
    print(f"正在处理文件: {target_filename}", file=sys.stderr)

# 处理FlowFile内容
for line in sys.stdin:
    print(line.strip())

3. session对象访问说明 #

通过ExecuteStreamCommand调用外部Python3脚本无法直接访问NiFi的session对象——session是NiFi Java进程内部的对象，外部脚本无法直接交互。若需session相关操作（如创建新FlowFile、将FlowFile转移至不同关系），有两种替代方案：

• 使用ExecuteScript组件搭配Jython 2.7，参考官方Cookbook的示例实现
• 通过NiFi的REST API间接操作，但该方式复杂度高，不适合常规FlowFile处理场景

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