如何通过NiFi处理器确认所有分片已完成处理（成功/失败）

针对你提出的“确认所有分片（无论成功/失败）完成处理”的需求，以下是两种基于NiFi原生处理器的可行方案：

方案一：使用Notify + Wait 处理器组合（推荐，原生无额外依赖） #

NiFi的Notify和Wait是专门用于协调一组任务完成状态的处理器，适配分片场景：

• 在SplitRecord处理器之后，将所有分片（包括后续处理成功/失败的分支）都连接到Notify处理器：
  • 设置Notification Identifier为 ${fragment.identifier}（该属性由SplitRecord自动生成，同一原始文件的所有分片共享唯一值）
  • 设置Count为 1，Total Count为 ${fragment.count}（总分片数，SplitRecord自动生成）
• 同时，从SplitRecord处理器单独分支一条流到Wait处理器：
  • 设置Wait Identifier为 ${fragment.identifier}
  • 设置Expected Count为 ${fragment.count}
• Wait处理器会阻塞等待，直到同一Notification Identifier下的Notify触发次数达到Total Count，此时即代表所有分片已完成处理（无论成功或失败），Wait的Success关系会触发后续的确认逻辑（比如日志记录、告警通知等）

关键注意：必须将ExecuteStreamCommand的Success和Failure两个关系都连接到Notify处理器，确保失败的分片也被计入完成计数。

方案二：使用DistributedMapCache 跟踪分片状态 #

如果需要更自定义的状态跟踪，可借助分布式缓存实现：

1. 在SplitRecord之后添加DistributedMapCachePut处理器：
   • 缓存键设为 ${fragment.identifier}
   • 缓存值设为 JSON 格式字符串：{"total": ${fragment.count}, "completed": 0}
2. 将ExecuteStreamCommand的Success和Failure关系都连接到DistributedMapCacheGet处理器，获取当前缓存的状态
3. 添加UpdateAttribute处理器，计算更新后的完成数：completed = ${completed} + 1
4. 用DistributedMapCachePut将更新后的状态写回缓存
5. 添加RouteOnAttribute处理器，设置路由规则：${completed:equals(${total})}，当条件满足时，触发“所有分片完成”的后续逻辑

关键注意：集群环境下需配置分布式缓存服务（如NiFi自带的DistributedMapCacheServer），确保状态在集群节点间共享。

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