NiFi中是否存在等待全部文件到齐后再写入HDFS的处理器

NiFi没有开箱即用的专属处理器实现「等待指定文件全部到齐再往下游流转」的能力，但可以通过官方自带的处理器组合轻松实现该需求，针对你提到的SFTP拉取5个文件后再写入HDFS的场景，具体配置方案如下：

所需核心处理器 #

• ListSFTP / FetchSFTP：负责从SFTP服务端拉取目标文件
• UpdateAttribute：给同批次文件打统一的聚合标识
• Notify + Wait：实现同批次文件的计数等待逻辑，是该方案的核心组件
• PutHDFS：负责将齐套的文件批量写入HDFS

详细配置步骤 #

1. 给同批次文件打统一标识
   通过UpdateAttribute处理器给属于同一批次的5个文件新增自定义属性batch_id，如果你的5个文件有统一的命名规则（比如前缀均为20240520_business_），可以直接用NiFi表达式提取公共前缀作为batch_id的值，保证同批次文件的batch_id完全一致。
2. 配置Notify计数处理器
   将FetchSFTP拉取成功的流文件路由到Notify处理器，核心配置如下：

• Signal Identifier（信号标识）填写${batch_id}，保证同批次文件的计数归到同一分组
• Counter Name（计数器名称）填写自定义值，比如arrived_file_num
• Delta Calculation（增量计算规则）选择「Increment」（累加模式）

3. 配置Wait等待处理器
   同样将FetchSFTP拉取成功的流文件路由到Wait处理器，核心配置如下：

• Signal Identifier（信号标识）和Notify保持一致，填写${batch_id}
• Wait Threshold（等待阈值）填写你需要的同批次文件总数，这里填固定值5
• Counter Name（计数器名称）和Notify保持一致，填写arrived_file_num
  配置完成后，当同一batch_id下的文件数累计到5个时，Wait会将该批次所有暂存的流文件全部释放到success关系。

4. 写入HDFS
   将Wait的success关系直接路由到PutHDFS，配置好HDFS的NameNode地址、写入路径、权限等参数即可，最终实现5个文件齐套后才批量写入HDFS的效果。

补充优化建议 #

• 如果需要处理超时场景，可以在Wait处理器中配置「Expiration Period」（过期时间），超时未齐套的流文件会被路由到timeout关系，你可以按需做告警或者重试逻辑
• 如果不同批次的文件总数不固定，可以提前拉取批次对应的预期文件总数，写入自定义属性total_file_num，再将Wait的阈值改成${total_file_num}即可适配动态数量的场景

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