DataWorks:For-each node

Built-in variables

Use the following built-in variables inside the loop body to access loop state and data:

These variables map directly to a standard for loop in Java:

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

Minimal example: If the upstream assignment node outputs a,b,c, the for-each node runs three iterations. In the second iteration, ${dag.foreach.current} is b, ${dag.offset} is 1, and ${dag.loopTimes} is 2.

Step 1: Configure the upstream assignment node

Create an assignment node that produces the result set the for-each node will iterate over.

1. In the workflow, create an assignment node (for example, assign) and place it upstream of the for-each node.

2. Double-click the assignment node and select a language. For example, use Python 2 to output a four-element array:

   print "10,20,30,40"

   The assignment node splits the last line of output by commas to form an array. This outputs [10, 20, 30, 40] to the downstream node.

3. The assignment node automatically generates an output parameter named outputs to represent its result set.

4. Save the assignment node.

Step 2: Bind the result set to the for-each node

1. Double-click the for-each node to open its internal canvas.

2. On the Scheduling settings panel, locate loopDataArray under Scheduling parameters and click Bind.

   

3. In the dialog box, set Value Source to the outputs parameter of the upstream assignment node (assign). This also establishes the dependency automatically.

Step 3: Build the loop body

1. In the for-each loop body canvas, click Create Inner Node and select a Shell node.

   In a production scenario, use any supported node type.

2. Double-click the Shell node and use built-in variables to access loop state:

   #!/bin/bash
   # Current iteration number (starts at 1)
   echo "Current loop number is: ${dag.loopTimes}"
   
   # Current data item
   echo "Current item is: ${dag.foreach.current}"

3. (Optional) On the Scheduling settings panel, configure Scheduling strategy:

   • Maximum number of cycles: Defaults to 128. Increase this value if your dataset has more than 128 items (maximum 1024).

   • Execution mode: Select Serial or Parallel. For this example, select Serial. If you select Parallel, the node runs up to 5 iterations concurrently by default (maximum 20).

4. Save the Shell node.

Step 4: Deploy, run, and verify

1. Return to the main workflow canvas and click Deploy on the toolbar to deploy the entire workflow to Operation Center.

2. Go to Operation Center > Task O&M > Periodic Task and start smoke testing on the assignment node (not the for-each node).

   Important

   Start smoke testing from the assignment node. The for-each node depends on the output of the assignment node — running it in isolation means loopDataArray has no input to iterate over.

3. After the test instance runs successfully, find the for-each node instance in the list. Right-click it and select View Inner Nodes.

   

4. In the inner node view, open the runtime log of any Shell node instance to verify the output for that iteration.

   

Scenario 1: One-dimensional array (Shell or Python output)

• Assignment node output: 2025-11-01,2025-11-02,2025-11-03

• Number of iterations: 3

• In the second iteration:

  • ${dag.foreach.current} = 2025-11-02

  • ${dag.loopTimes} = 2

Scenario 2: Two-dimensional array (SQL output)

• Assignment node output (MaxCompute SQL):

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

• Number of iterations: 2

• In the second iteration:

  • ${dag.foreach.current} = 102,shanghai

  • ${dag.loopTimes} = 2

  • ${dag.foreach.current[0]} = 102

  • ${dag.foreach.current[1]} = shanghai

Background

Assume you are a data developer at a large internet company. You process daily user behavioral data from three lines of business: e-commerce, finance, and logistics. More lines may be added in the future. Each day, you run the same aggregation — calculating page views (PV) per user — on each line's behavioral logs and write results into a shared aggregate table.

Source tables (DWD layer), partitioned by day (dt):

• dwd_user_behavior_ecom_d

• dwd_user_behavior_finance_d

• dwd_user_behavior_logistics_d

• dwd_user_behavior_${line-of-business}_d (for future lines)

Destination table (DWS layer), partitioned by biz_line and dt:

• dws_user_summary_d

Using a for-each node, you maintain one set of processing logic. At runtime, the assignment node determines the list of lines of business, and the for-each node iterates over it automatically.

Data preparation

1. Associate a MaxCompute computing resource with the workspace.

2. Go to Data Studio and create a MaxCompute SQL node.

3. Create the source tables: add the following code, select it, and run it.

   -- E-commerce user behavior table
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_ecom_d (
       user_id     STRING COMMENT 'User ID',
       action_type STRING COMMENT 'Behavior type',
       event_time  BIGINT COMMENT 'Millisecond-level UNIX timestamp of the event occurrence'
   )
   COMMENT 'Details of e-commerce user behavioral logs'
   PARTITIONED BY (dt STRING COMMENT 'Date partition in yyyymmdd format');
   
   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
   -- Verify that the e-commerce user behavior table is created.
   SELECT * FROM dwd_user_behavior_ecom_d where dt='20251010';
   
   -- Finance user behavior table
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_finance_d (
       user_id     STRING COMMENT 'User ID',
       action_type STRING COMMENT 'Behavior type',
       event_time  BIGINT COMMENT 'Millisecond-level UNIX timestamp of the event occurrence'
   )
   COMMENT 'Details of finance user behavioral logs'
   PARTITIONED BY (dt STRING COMMENT 'Date partition in yyyymmdd format');
   
   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
   -- Verify that the finance user behavior table is created.
   SELECT * FROM dwd_user_behavior_finance_d where dt='20251010';
   
   -- Logistics user behavior table
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_logistics_d (
       user_id     STRING COMMENT 'User ID',
       action_type STRING COMMENT 'Behavior type',
       event_time  BIGINT COMMENT 'Millisecond-level UNIX timestamp of the event occurrence'
   )
   COMMENT 'Details of logistics user behavioral logs'
   PARTITIONED BY (dt STRING COMMENT 'Date partition in yyyymmdd format');
   
   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
   -- Verify that the logistics user behavior table is created.
   SELECT * FROM dwd_user_behavior_logistics_d where dt='20251010';

4. Create the destination table: add the following code, select it, and run it.

   Important

   If the workspace uses standard mode, deploy this node to the production environment and perform a data backfill.

   CREATE TABLE IF NOT EXISTS dws_user_summary_d (
       user_id     STRING COMMENT 'User ID',
       pv          BIGINT COMMENT 'Daily popularity'
   )
   COMMENT 'Daily user popularity aggregate table'
   PARTITIONED BY (
       dt       STRING COMMENT 'Date partition in yyyymmdd format',
       biz_line STRING COMMENT 'Line-of-business partition, such as ecom, finance, logistics'
   );

Workflow implementation

1. Create a workflow. In the Scheduling pane, set the scheduling parameter bizdate to the previous day: $[yyyymmdd-1].

   

2. Create an assignment node named get_biz_list in the workflow. Write the following MaxCompute SQL to output the list of lines of business:

   -- Output all lines of business to process
   SELECT 'ecom' AS biz_line
   UNION ALL
   SELECT 'finance' AS biz_line
   UNION ALL
   SELECT 'logistics' AS biz_line;

3. Configure the for-each node:

   • On the workflow canvas, create a for-each node downstream of get_biz_list.

   • On the Scheduling tab, under Scheduling Parameters > Script Parameters, bind loopDataArray to the outputs parameter of get_biz_list.

   • In the for-each loop body, click Create Internal Node and create a MaxCompute SQL node with the following processing logic:

     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;

     Note: This script runs once per iteration. ${dag.foreach.current} is replaced at runtime with the current line-of-business name: ecom, finance, or logistics.

4. Add a verification node: on the workflow canvas, create a MaxCompute SQL node downstream of the for-each node with the following code:

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

Deploy and run

Deploy the workflow to the production environment. In Operation Center, go to Auto Triggered Node O&M > Auto Triggered Nodes, find the workflow, run a test, and select 20251010 as the data timestamp.

After the run completes, open the runtime log in the test instance. The expected output of the final node is:

Why this approach scales: To add a new line of business, add one UNION ALL SELECT line in get_biz_list. The for-each node picks it up automatically — no changes to the processing logic required.