本文介绍 For Each 类型的 Activity 配置，用于在 Pipeline 中实现集合遍历。

概述 #

For Each Activity 对集合中的每个元素执行一组内部 Activity，适用于需要批量处理多个同类对象的场景。
适用场景：

• 按地区/租户/品类分别执行相同的 ETL 逻辑。
• 遍历文件列表逐个处理。
• 动态数量的并行任务。

配置示例 #

- name: process_regions
  type: for_each
  items: '["cn-beijing", "cn-shanghai", "cn-guangzhou", "cn-shenzhen"]'
  concurrency: 3
  activities:
    - name: etl_per_region
      type: sql
      source: WORKSPACE
      path: /Workspace/Users/zhang3/sql/region_etl.sql
      engineType: emr_serverless_spark
      engineQueue: default
      parameterValues:
        region_id: "{{items.each}}"
      position:
        x: "100"
        y: "50"
  position:
    x: "200"
    y: "100"

字段说明 #

items 定义 #

静态数组 #

直接以 JSON 数组格式写入遍历元素：

items: '["cn-beijing", "cn-shanghai", "cn-guangzhou"]'

动态数组 #

引用前序 Activity 的输出作为遍历集合：

items: "{{activities.get_regions.output.rows}}"

前序 Activity 的输出需要是 JSON 数组格式。

参数引用 #

引用 Pipeline 参数或变量：

items: "{{pipeline.parameters.region_list}}"

内部 Activity #

activities 数组定义每次迭代要执行的 Activity 列表。内部 Activity 的配置方式与普通 Activity 相同，但额外支持 {{items.each}} 引用当前迭代元素。

引用当前元素 #

activities:
  - name: etl_per_region
    type: sql
    source: WORKSPACE
    path: /Workspace/Users/zhang3/sql/region_etl.sql
    engineType: emr_serverless_spark
    engineQueue: default
    parameterValues:
      region_id: "{{items.each}}"          # 当前迭代元素

多步骤迭代 #

每次迭代可包含多个按 DAG 编排的 Activity：

- name: process_each_region
  type: for_each
  items: '["cn-beijing", "cn-shanghai"]'
  concurrency: 2
  activities:
    - name: extract
      type: sql
      source: WORKSPACE
      path: /Workspace/Users/zhang3/sql/extract_by_region.sql
      engineType: emr_serverless_spark
      engineQueue: default
      parameterValues:
        region: "{{items.each}}"
      position:
        x: "100"
        y: "50"

    - name: transform
      type: sql
      source: WORKSPACE
      path: /Workspace/Users/zhang3/sql/transform_by_region.sql
      engineType: emr_serverless_spark
      engineQueue: default
      parameterValues:
        region: "{{items.each}}"
      dependsOn:
        activities:
          - extract
      position:
        x: "300"
        y: "50"

并发控制 #

通过 concurrency 控制同时运行的迭代数量：

# 最多 3 个地区并行处理
concurrency: 3

执行流程 #

For Each 节点开始
  → 解析 items 得到元素列表 [e1, e2, e3, e4]
    → 按 concurrency 控制并行度
      → 迭代 1 (e1): 执行 activities → 完成
      → 迭代 2 (e2): 执行 activities → 完成
      → 迭代 3 (e3): 执行 activities → 完成
      → 迭代 4 (e4): 执行 activities → 完成
  → 所有迭代完成 → For Each 节点 SUCCEEDED

失败行为：

• 某次迭代中的 Activity 执行失败时，该迭代标记为失败。
• 已启动的其他迭代继续执行（不会提前终止）。
• 所有迭代完成后，如果有任意迭代失败，For Each 节点整体标记为 FAILED。

使用建议 #