for-each节点

内置变量

在 for-each 循环体内部的节点中，使用以下内置变量来获取循环状态和数据：

for(int i=0;i<data.length;i++) {
   print(data[i]);
}

当上游输出为二维数组时（如 SQL 查询结果），还可使用以下方式精确取值：

场景1：处理一维数组（Shell/Python 输出）

• 赋值节点输出：2025-11-01,2025-11-02,2025-11-03

• 遍历次数：3 次。

• 第2次循环时：

  • ${dag.foreach.current} 的值为 2025-11-02。

  • ${dag.loopTimes} 的值为 2。

场景2：处理二维数组（SQL 输出）

• 赋值节点 (MaxCompute SQL) 输出：

  +-----+----------+
  | id  | city     |
  +-----+----------+
  | 101 | beijing  |
  | 102 | shanghai |
  +-----+----------+

• 遍历次数：2 次。

• 第2次循环时：

  • ${dag.foreach.current} 的值为 102,shanghai。

  • ${dag.loopTimes} 的值为 2。

  • ${dag.foreach.current[0]} 的值为 102。

  • ${dag.foreach.current[1]} 的值为 shanghai。

业务背景

假设您是一家综合性互联网公司的数据开发工程师，负责处理三个核心业务线的数据：电商（ecom）、金融（finance）和物流（logistics），且后续存在增加业务线的可能。您需要每天对这三个业务线的用户行为日志执行相同的聚合逻辑，计算每个用户的日活跃度（PV），并将结果存入统一的汇总表。

• 上游源表（DWD层）：

  • dwd_user_behavior_ecom_d：电商用户行为表。

  • dwd_user_behavior_finance_d ：金融用户行为表。

  • dwd_user_behavior_logistics_d ：物流用户行为表。

  • dwd_user_behavior_${业务线}_d ：后续更多可能的业务线用户行为表。

  • 这些表结构相同，都按天分区（dt）。

• 下游目标表（DWS）：

  • dws_user_summary_d ：用户汇总表。

  • 该表按业务线（biz_line）和天（dt）双重分区，用于统一存储所有业务线的聚合结果。

若为每个业务线创建独立的任务，维护成本高且容易出错。使用 for-each 节点后，只需维护一份处理逻辑，系统会自动遍历所有业务线完成计算。

数据准备

首先，创建示例表并插入测试数据（以业务日期 20251010 为例）：

1. 为工作空间绑定MaxCompute计算资源。

2. 进入Data Studio数据开发，创建MaxCompute SQL节点。

3. 创建源表（DWD层）：在MaxCompute SQL节点添加如下代码，并选中运行。

   -- 电商用户行为表
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_ecom_d (
       user_id     STRING COMMENT '用户ID',
       action_type STRING COMMENT '行为类型',
       event_time  BIGINT COMMENT '事件发生的毫秒级Unix时间戳'
   ) 
   COMMENT '电商用户行为日志明细表'
   PARTITIONED BY (dt STRING COMMENT '日期分区，格式 yyyymmdd');
   
   INSERT OVERWRITE TABLE dwd_user_behavior_ecom_d PARTITION (dt='20251010') VALUES
   ('user001', 'click',        1760004060000), -- 2025-10-10 10:01:00.000
   ('user002', 'browse',       1760004150000), -- 2025-10-10 10:02:30.000
   ('user001', 'add_to_cart',  1760004300000); -- 2025-10-10 10:05:00.000
   -- 验证电商用户行为表创建成功
   SELECT * FROM dwd_user_behavior_ecom_d where dt='20251010';
   
   -- 金融用户行为表
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_finance_d (
       user_id     STRING COMMENT '用户ID',
       action_type STRING COMMENT '行为类型',
       event_time  BIGINT COMMENT '事件发生的毫秒级Unix时间戳'
   ) 
   COMMENT '金融用户行为日志明细表'
   PARTITIONED BY (dt STRING COMMENT '日期分区，格式 yyyymmdd');
   
   INSERT OVERWRITE TABLE dwd_user_behavior_finance_d PARTITION (dt='20251010') VALUES
   ('user003', 'open_app',      1760020200000), -- 2025-10-10 14:30:00.000
   ('user003', 'transfer',      1760020215000), -- 2025-10-10 14:30:15.000
   ('user003', 'check_balance', 1760020245000), -- 2025-10-10 14:30:45.000
   ('user004', 'open_app',      1760020300000); -- 2025-10-10 14:31:40.000
   -- 验证金融用户行为表创建成功
   SELECT * FROM dwd_user_behavior_finance_d where dt='20251010';
   
   -- 物流用户行为表
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_logistics_d (
       user_id     STRING COMMENT '用户ID',
       action_type STRING COMMENT '行为类型',
       event_time  BIGINT COMMENT '事件发生的毫秒级Unix时间戳'
   ) 
   COMMENT '物流用户行为日志明细表'
   PARTITIONED BY (dt STRING COMMENT '日期分区，格式 yyyymmdd');
   
   INSERT OVERWRITE TABLE dwd_user_behavior_logistics_d PARTITION (dt='20251010') VALUES
   ('user001', 'check_status',    1760032800000), -- 2025-10-10 18:00:00.000
   ('user005', 'schedule_pickup', 1760032920000); -- 2025-10-10 18:02:00.000
   
   -- 验证物流用户行为表创建成功
   SELECT * FROM dwd_user_behavior_logistics_d where dt='20251010';

4. 创建目标表（DWS层）：在MaxCompute SQL节点上添加如下代码，并选中运行。

   CREATE TABLE IF NOT EXISTS dws_user_summary_d (
       user_id     STRING COMMENT '用户ID',
       pv          BIGINT COMMENT '日活跃度',
   ) 
   COMMENT '用户日活跃度汇总表'
   PARTITIONED BY (
       dt           STRING COMMENT '日期分区，格式 yyyymmdd',
       biz_line     STRING COMMENT '业务线分区，如 ecom, finance, logistics'
   );

   重要

   若工作空间使用标准环境，需要将此节点发布至生产环境，并执行补数据。

工作流实现

1. 创建一个工作流，并在右侧调度参数设置调度参数bizdate为上一天$[yyyymmdd-1]。

   

2. 在工作流中，创建名为get_biz_list的赋值节点，并使用MaxCompute SQL语言编写如下代码。该节点输出需要处理的业务线列表：

   -- 输出所有需要处理的业务线
   SELECT 'ecom' AS biz_line
   UNION ALL
   SELECT 'finance' AS biz_line
   UNION ALL
   SELECT 'logistics' AS biz_line;

3. 配置 for-each 节点

   • 回到工作流界面，给赋值节点get_biz_list创建一个下游的for-each节点。

   • 进入for-each节点设置界面，在右侧调度配置的调度参数 > 脚本参数，将 loopDataArray 参数绑定到get_biz_list节点的outputs。

     

   • 在for-each节点循环体内单击新建内部节点，创建 MaxCompute SQL节点，编写循环体内的处理逻辑。

     说明

     • 该脚本由 for-each 节点驱动，会针对每个业务线执行一次。

     • 内置变量 ${dag.foreach.current} 在每次运行时，会动态地替换为当前的业务线名称。预期的迭代值为：'ecom', 'finance', 'logistics'。

     SET odps.sql.allow.dynamic.partition=true;
     
     INSERT OVERWRITE TABLE dws_user_summary_d PARTITION (dt='${bizdate}', biz_line)
     SELECT
         user_id,
         COUNT(*) AS pv,
         '${dag.foreach.current}' AS biz_line
     FROM
         dwd_user_behavior_${dag.foreach.current}_d
     WHERE
         dt = '${bizdate}'
     GROUP BY
         user_id;

4. 添加验证节点

   回到工作流，在for-each节点单击新建下游，创建一个MaxCompute SQL节点并添加如下代码。

   SELECT * FROM dws_user_summary_d WHERE dt='20251010' ORDER BY biz_line, user_id;

发布和运行结果

将工作流发布至生产环境，前往运维中心的周期任务运维 > 周期任务，找到目标工作流并执行冒烟测试，选择业务日期为'20251010'。

完成运行后，在测试实例中查看运行日志。最终节点预期输出为：

方案优势

• 高扩展性：新增业务线时，只需在赋值节点添加一行 SQL，无需修改处理逻辑。

• 易维护：所有业务线共享同一套处理逻辑，修改一处即可生效于全部。