NiFi如何等待所有处理器执行完成后合并同原始CSV的全部FlowFile

前置准备：全局唯一标识绑定 #

• 在GetFile处理器后新增UpdateAttribute处理器，新增两个属性：
  • original.csv.id：值设置为${UUID()}，作为当前原始CSV的全局唯一标识
  • original.csv.filename：值设置为${filename}，用于后续输出文件命名
• 所有后续的拆分、转换、路由处理器，均需开启「继承父FlowFile全部属性」的配置，确保所有从该CSV衍生的子FlowFile，都携带以上两个属性

核心逻辑：基于原始行完成状态的触发机制 #

由于仅原始CSV拆分后的行总数是固定值，不受后续行内拆分逻辑影响，因此用原始行的处理完成状态作为整个CSV处理完成的判断依据：

1. 原始行计数标记：在SplitJson处理器后新增UpdateAttribute，新增original.total.rows属性，值设置为${fragment.count}，该值为当前CSV拆分出的原始行总数，固定不变
2. 单原始行处理完成通知：在每个转换分支的末尾（所有子FlowFile生成完成后，流入Funnel之前），新增Notify处理器，配置如下：
   • Notification Topic Name：row-processed-${original.csv.id}
   • Attribute to Cache：填入fragment.identifier（即SplitJson生成的原始行唯一标识）
3. 全量行完成等待触发：在SplitJson处理器后并行新增Wait处理器，配置如下：
   • Wait Target：${original.total.rows}
   • Notification Topic Name：row-processed-${original.csv.id}
   • Attribute Buffer Size：设置为大于业务中单个CSV最大行数的值即可
   • 等待条件满足后输出的FlowFile，代表当前CSV的所有衍生FlowFile已全部流入Funnel，给该FlowFile新增merge.trigger = true属性后汇入Funnel

MergeContent配置 #

• 合并策略选择「Use Correlation Attribute」，Correlation Attribute Name填入original.csv.id，仅合并属于同一个原始CSV的FlowFile
• Minimum Number of Entries设为1，Maximum Number of Entries设为大于业务中单个CSV衍生的最大FlowFile数的值
• 新增触发规则：当同一original.csv.id分组中存在merge.trigger = true属性的FlowFile时，立即触发该分组的合并，合并时过滤掉带merge.trigger属性的触发用FlowFile

注意事项 #

• 若使用NiFi集群，需将Notify和Wait处理器关联的缓存控制器配置为集群统一分布式缓存，禁止使用本地缓存
• 可在Funnel前新增RouteOnAttribute处理器，过滤空内容的FlowFile，避免合并冗余内容
• 输出文件名可通过original.csv.filename属性自定义，保留原始文件关联关系

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