Apache NiFi实现无限循环、嵌套逻辑及休眠功能的技术咨询

需要在Apache NiFi中实现以下循环逻辑：

while(true) {
  prj_result = invoke_prj_Rest_EndPoint();
  prjs_list = prjs_list["prjs"];
  for prj in prjs_list:
       emp_response = invoke_emp_rest_Endpoint(prj["id"]);
       next_cursor = emp_response["meta"]["next_cursor"];
       while next_cursor:
         sleep(10 secs);
         emps = emp_response["emps"];
         for emp in emps:
             emp["mod_time"] = now();
         loadEmpDataToS3()
  sleep(5Mins)
}

已有初步实现方案，针对以下疑问寻求优化方向：

疑问解答 #

1. 无限循环（while(true)）的更优实现 #

不用初始化参数判断的方式，推荐用闭环路由+原生处理器实现：

• 用GenerateFlowFile生成一个初始触发的种子FlowFile，调度策略可设为立即执行或按需配置。
• 在整个流程的最后（完成一轮prj和emp数据处理后），添加RouteOnAttribute处理器，设置一个永远为真的路由规则（比如loop_again = true），将该路由分支直接连回调用prj接口的处理器，形成闭环。
• 如果需要控制每轮循环的间隔（比如5分钟），在闭环路由前添加Wait处理器，设置等待时长为5分钟即可。这种方式无需维护额外参数变量，逻辑更直观，也更符合NiFi的组件化设计。

2. 休眠功能的最佳实现 #

优先使用NiFi原生的Wait处理器：

• 支持固定时长等待（直接配置10秒、5分钟），也能基于FlowFile属性动态设置等待时间，灵活性拉满。
• 相比用ExecuteScript写Thread.sleep()代码，Wait是官方维护的原生组件，稳定性更高，无需额外代码维护，监控和调试也更方便。
• 场景适配：分页循环里的10秒休眠，就在获取下一页请求前插入Wait处理器；每轮大循环的5分钟休眠，放在整个流程末尾、闭环路由前即可。

3. 遍历prjs_list和分页循环的实现 #

遍历prjs_list（for循环）

用SplitJson+EvaluateJsonPath组合实现：

1. 调用prj接口后，用EvaluateJsonPath提取响应中的prjs数组，可将其保存到FlowFile内容或属性中。
2. 添加SplitJson处理器，配置JsonPath为$..prjs[*]，将数组中的每个prj对象拆分为独立的FlowFile。
3. 每个拆分后的FlowFile单独进入后续的emp接口调用流程，自然实现遍历效果。

基于next_cursor的分页循环

用RouteOnAttribute+InvokeHTTP+Wait组合实现：

1. 调用emp接口后，用EvaluateJsonPath提取next_cursor和emps数据，将next_cursor存入FlowFile属性（比如命名为next_cursor）。
2. 添加RouteOnAttribute处理器，设置路由规则：
   • 当${next_cursor:notEmpty()}为真时：先通过Wait处理器等待10秒，再将FlowFile路由回InvokeHTTP处理器（调用emp接口时，把next_cursor作为请求参数传入），继续获取下一页数据。
   • 当${next_cursor:isEmpty()}为真时：将FlowFile路由到后续处理逻辑。
3. 处理当前页emps数据时：用EvaluateJsonPath提取emps数组，通过SplitJson拆分每个emp对象，再用UpdateAttribute添加mod_time属性（值设为${now()}），最后用PutS3Object将数据上传到S3。

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