如何在NiFi中将ExecuteStreamCommand输出流结果写入原始流文件

ExecuteStreamCommand 处理器默认行为是将外部脚本写入stdout的内容作为output stream关系的FlowFile内容，不会自动保留原始输入FlowFile的内容，要实现输出结果追加到原始内容的需求，可根据场景选以下两种方案：

方案1：脚本侧直接拼接（性能最优，优先推荐） #

该方案不需要额外配置下游处理器，磁盘IO和内存开销最低，适合可以修改Python脚本的场景：

• 配置ExecuteStreamCommand的Input Handling参数为stdin，此时原始FlowFile的内容会作为标准输入流传入Python脚本
• Python脚本内先读取stdin拿到原始内容，完成业务逻辑处理后，将原始内容和处理结果按要求拼接，统一输出到stdout即可
• 参考脚本实现：

import sys

# 文本场景读取原始内容
# original_content = sys.stdin.read()
# 二进制内容（文件、压缩包、媒体资源等）用二进制模式读取，避免编码损坏
original_content = sys.stdin.buffer.read()

# 替换为你实际的业务处理逻辑，生成对应结果
# 文本场景直接用字符串，二进制场景用字节类型
process_result = b"Python脚本实际处理生成的输出内容"

# 按需要的格式拼接原始内容和处理结果，输出到stdout
# 这里示例用换行符分隔两部分内容，可根据需求调整
final_content = original_content + b"\n" + process_result
# 二进制场景用buffer.write输出，文本场景直接print即可
sys.stdout.buffer.write(final_content)

• 注意事项：脚本的调试、日志类内容请输出到stderr，这部分内容会被NiFi记录到处理器日志中，不会混入FlowFile内容；不要把业务结果打印到stdout之外的通道，否则会丢失输出。

方案2：流程侧合并（适合无法修改原有Python脚本的场景） #

如果已经有写好的Python脚本不方便改动，可以通过NiFi处理器串联完成内容拼接：

• ExecuteStreamCommand有两个核心导出关系：Original关系会导出完全未修改的原始输入FlowFile，output stream关系会导出脚本stdout生成的结果FlowFile，两个同源的FlowFile会携带相同的fragment.identifier属性用于关联
• 流程连接逻辑：将ExecuteStreamCommand的Original、output stream两个关系同时连接到下游的MergeContent处理器
• 配置MergeContent参数：
  • 合并策略选择Defragment
  • 分隔符按需求设置，比如需要换行分隔两部分内容就填\n
  • 最小合并条目数、最大合并条目数都设置为2，处理器会自动将同个fragment.identifier的两个FlowFile按顺序合并
• 注意事项：需要合理设置MergeContent的分片过期时间，避免异常场景下残留的FlowFile长期占用堆内存。

踩坑提醒：不要为了省事把脚本输出写入FlowFile属性，FlowFile属性是存储在内存中的，输出内容过大会直接触发NiFi堆内存溢出，大体积内容必须存放在FlowFile的内容流中。

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