For-each node

Use cases

A for-each node enables parameterized execution when you need to apply the same analysis or processing logic to different business units, product lines, or configuration items. For example, if your company has multiple product lines and you need to generate a separate daily report for each, the processing logic is identical; only the target data differs.

Similar to a for loop in a programming language, a for-each node iterates over a list, such as table names, partition names, or filenames, and executes a predefined sub-workflow for each item.

Usage notes

• Version requirements: Available only in DataWorks Standard Edition and later.

• Permissions: Your RAM account must be added to the target workspace and assigned the developer or workspace administrator role. For more information, see Add members to a workspace.

How it works

The for-each node acts as a container that encapsulates a customizable sub-workflow, known as the loop body. It works as follows:

1. Data input: The for-each node depends on an upstream assignment node or other assignable node (such as an EMR Hive node). It retrieves the array-formatted result set by binding to the loopDataArray parameter.

2. Loop execution: When the node starts, it iterates through each element in the result set in order. For each element, it fully executes the inner loop body once, from the Start node to the End node.

   Note

   The Start and End nodes are not editable. They only mark the beginning and end of the loop body.

3. Data passing: During each iteration, the value of the current element is passed to the nodes inside the loop body via built-in variables. The internal business nodes use ${dag.foreach.current} to access the data item being processed.

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

Limitations

• Execution mechanism: The loop supports both serial execution and parallel execution. You can choose parallel execution when the iterations are independent of each other.

• Loop limit: The default maximum number of loops is 128, which can be adjusted up to 1024.

• Debugging constraints: You cannot run a for-each node directly in Data Studio. You must deploy the task and then test it in Operation Center by using the smoke testing feature.

• Execution constraints: A for-each node cannot be run in isolation. This includes smoke testing, backfill, and manual runs.

• Flow control in the loop body: If you use a branch node inside a for-each loop body, you must ensure all branches eventually converge at a single merge node before connecting to the End node. This guarantees the logical integrity of the loop body.

• Rerun constraints: After a node is deployed, an automatic rerun on failure resumes from the point of failure. However, a manual rerun triggers a complete rerun of the entire for-each node.

Procedure

This procedure uses an assignment node as the upstream node and a Shell node inside the loop body to print the results. It walks through configuring a complete for-each task:

1. Prepare the upstream data (configure an assignment node)

   Create and configure an assignment node to provide an iterable result set for the downstream for-each node.

   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 Python 2 environment. For example, use Python 2 to output an array with four elements:

      The node outputs [10,20,30,40] to downstream nodes by automatically splitting the last output line into an array at each comma.

      print "10,20,30,40"

   3. The assignment node automatically generates an output parameter named outputs, which represents its result set.

   4. Save the assignment node.

2. Configure the for-each node to consume data

   Configure the for-each node to receive the upstream data and use it within its loop body.

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

   2. In the Scheduling panel on the right, find the loopDataArray parameter under Scheduling Parameters and click Bind.

      Select the outputs parameter of the assign node to create the binding. After the binding is complete, the value of the loopDataArray parameter reflects its bound status.

   3. In the dialog box that appears, set the Value Source to the upstream assignment node (assign) and select its outputs parameter. This action automatically creates a dependency between the two nodes.

   4. In the for-each loop body, click Create Internal Node and create a Shell node.

      In a real-world scenario, you can configure any type of node.

   5. Double-click the new Shell node and use built-in variables in the code to retrieve and print information about the loop:

      #!/bin/bash
      # Use ${dag.loopTimes} to get the current loop count
      echo "Current loop number is: ${dag.loopTimes}"
      # Use ${dag.foreach.current} to get the data item for the current iteration
      echo "Current item is: ${dag.foreach.current}"

   6. (Optional) In the Scheduling Settings panel on the right, configure properties under Scheduling Policy.

      • Maximum Number of Loops: The default is 128, and the maximum is 1024.

        Important

        This parameter determines the maximum number of iterations for the loop body. If the number of upstream data items is large, increase this value to ensure all items are processed.

      • Execute Policy: Select Serial for this example.

        • Serial: Runs iterations sequentially.

        • Parallel: Runs loop iterations concurrently to improve task efficiency. In Parallel mode, if one iteration fails, it does not affect other iterations. The scheduler attempts to run all iterations to completion. The default concurrency is 5, and the maximum is 20.

   7. Save the Shell node.

3. Deploy, run, and verify

   Deploy the workflow to Operation Center for execution and verify the results of the for-each node.

   1. Return to the main workflow canvas and click the Deploy button on the toolbar to publish the entire workflow.

   2. Go to Node O&M > Auto Triggered Task O&M > Auto Triggered Task and perform a smoke test on the target workflow.

      Important

      Do not perform a smoke test on the for-each node individually. Because the for-each node depends on the output of the upstream assignment node, you must start the test from the assignment node to ensure the data lineage is complete.

   3. After the test instance runs successfully, find the for-each node instance in the list, open it, and right-click to select View Internal Nodes.

   4. In the internal node view, check the Shell node instances generated by each loop. Open the running log of any instance to view the output for that iteration and verify that the output is correct.

      The left panel shows that all four loop iterations are complete. The running log for the fourth iteration outputs Current loop number is: 4 and Current item is: 40, and the Shell command exits with code 0, indicating successful execution.

Use case: Process different data formats

Scenario: Batch process partition table data across multiple business lines

This example demonstrates how to use an assignment node and a for-each node to batch process user behavior data across multiple business lines, automating data processing with a single set of logic that serves multiple product lines.

FAQ

• Q: Why can't I run a for-each node directly in Data Studio to test it?

  A: This is by design. The node requires a full scheduling environment to resolve the node context and its dependencies, so it does not support direct execution in Data Studio. You must deploy the task to Operation Center and test it by using backfill or triggering a scheduled run.

• Q: Why does a smoke test on an individual for-each node fail or do nothing?

  A: The loop data for a for-each node comes from its loopDataArray input parameter, which must be bound to the outputs parameter of an upstream assignment node. If you run the for-each node by itself, it will either fail or be skipped because it cannot receive an input result set.

• Q: Why does my loop run only once? 

  A: This usually happens because the output from the upstream assignment node is parsed as a single element. Check your output:

  • 1. Is it a single string without a delimiter?

  • 2. If you expect to iterate over multiple items, ensure they are separated by commas (,). For example, 'item1,item2,item3' results in three loops, whereas 'item1 item2 item3' results in only one.