NiFi流程中如何通知Merge Processor在所有流文件保存完成后启动合并操作？

针对你这个需求，NiFi有几种原生且可靠的实现方式，我给你拆解两个最常用的方案：

方案一：使用Wait/Notify Processor组合（推荐） #

这是NiFi处理“等待批量任务完成再触发后续操作”的标准模式，步骤清晰且可控：

1. 统计待处理的流文件总数

   • 在从Elasticsearch获取数据并拆分/处理得到单个流文件后，添加一个AttributeRoutedCount Processor。配置Count Attribute Name为total_files，如果是动态批次，可以先用SplitContent生成的fragment.count属性，或者通过ExecuteScript计算总数后存入流文件属性。
   • 用DistributedMapCachePut把这个总计数存入分布式缓存，键名设为batch_${uuid}_total（用唯一批次ID区分不同任务，避免混淆）。

2. 单个文件保存完成后更新计数，触发全局完成通知

   • 在保存流文件的最后一步（比如PutFile）之后，添加Notify Processor，配置Notification Identifier为single_file_saved，同时传递批次ID属性。
   • 紧接着用DistributedMapCacheIncrement把缓存中的批次总计数减1。当计数减至0时，触发另一个Notify Processor，设置Notification Identifier为all_files_saved，并传递同一个批次ID。

3. Wait Processor监听完成通知，启动合并

   • 添加Wait Processor，配置它监听all_files_saved这个通知ID，同时匹配对应的批次ID。当收到这个全局完成通知时，Wait会释放一个触发流文件，直接连接到你的MergeRecord（推荐用于结构化CSV合并）或MergeContent Processor，启动合并操作。

方案二：利用Process Group的完成策略（简洁版） #

如果你的取数、处理、保存流程可以封装为独立单元，这个方法更省心：

1. 将从Elasticsearch取数、处理到保存的所有Processor，打包到一个Process Group中。
2. 编辑Process Group的配置，找到Completion Strategy选项，设置为All Tasks Completed。这个策略会等待组内所有流文件处理完成（无待执行任务）后，触发一个完成事件。
3. 在Process Group外部添加ListenProcessGroupStatus Processor，配置它监听该组的COMPLETED状态。当收到完成状态时，这个Processor会生成一个触发流文件，连接到Merge Processor启动合并。

额外提示 #

• 处理超大批次任务时优先选方案一，通过批次ID可以精准隔离不同任务的通知，避免交叉干扰。
• 合并CSV推荐用MergeRecord，记得配置对应的CSV格式Record Reader和Record Writer，确保合并后的文件结构符合预期。

内容的提问来源于stack exchange，提问作者Chandan Basant Tiwari