Apache NiFi中PutFile操作完成后执行Python脚本的实现方案咨询

下面提供两种经过验证的可行方案，可解决你之前遇到的配置问题：

方案1：使用Defragment模式MergeContent（配置更简单） #

• 第一步：配置UnpackContent组件，格式选择ZIP，解压后每个子文件会自动携带三个核心属性：fragment.identifier（同个压缩包内所有文件值一致）、fragment.count（该压缩包解压出的总文件数）、fragment.index（当前文件的序号）
• 第二步：连接UnpackContent的success关系到PutFile组件，配置PutFile的目标目录为指定写入路径，所有解压文件写入完成后会走success路由，此时success流出的FlowFile依然保留上述三个fragment属性
• 第三步：连接PutFile的success关系到MergeContent组件，合并策略直接选择「Defragment」，该模式会自动根据fragment.identifier分组、等待同一组的fragment.count个文件全部到达后自动合并，不需要手动设置最小/最大条目数，完全解决你之前无法将最小条目设为${fragment.count}的问题
• 第四步：合并完成后生成的单个FlowFile即代表该压缩包的所有文件已全部写入完成，直接连接MergeContent的success关系到ExecuteScript或ExecuteProcess组件，配置运行指定Python脚本即可。如果使用ExecuteProcess，Command填python3，Command Arguments填你的脚本绝对路径即可。

方案2：修正Notify/Wait配置（性能开销更低） #

如果你不需要合并所有文件内容，仅需要触发执行脚本，可以修正之前的Notify/Wait配置实现需求：

• 解压和写文件的前两步和方案1一致，将PutFile的success关系连接到Notify组件
• Notify组件配置：
  • 释放信号的标识符设置为${fragment.identifier}，保证同个压缩包的信号统一
  • 「信号计数属性」填写fragment.count，代表需要累计收到多少个同标识符的FlowFile才释放信号
  • 「到达计数时删除信号」选择是
• 将UnpackContent的original关系（即原压缩包对应的FlowFile）先连接到UpdateAttribute组件，新增属性fragment.identifier值为${fragment.identifier}，再连接到Wait组件
• Wait组件配置：
  • 等待的信号标识符设置为${fragment.identifier}，和Notify侧配置一致
  • 「等待目标计数」填写${fragment.count}
  • 等到信号释放后，Wait的success关系流出的FlowFile即可触发Python脚本执行，该方案不需要合并文件内容，大文件场景下性能更好。

注意事项 #

• 建议添加RouteOnAttribute组件提前过滤空压缩包，避免计数为0导致流程卡住
• 需处理PutFile的failure路由，不要让写入失败的文件进入Notify或MergeContent，避免计数错误
• 如果Python脚本需要用到目标目录、压缩包名等参数，可通过FlowFile属性传递给脚本作为入参。

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