概述

Execute Pipeline Activity 实现子 Pipeline 调用，将复杂的工作流拆分为多个可复用的 Pipeline 模块，由父 Pipeline 统一编排。
适用场景：

复用已有的 Pipeline 逻辑（如通用的数据质量检查 Pipeline）。
拆分大型 Pipeline 降低复杂度。
动态传参执行同一个 Pipeline 的不同业务场景。

配置示例

- name: run_quality_check
  type: execute_pipeline
  pipelineName: data_quality_check
  pipelineId: 456
  pipelineParameterValues:
    biz_date: "${date}"
    check_level: "strict"
    target_table: "dw.order_summary"
  waitingSuccess: true
  bizDate: "${date}"
  retryPolicy:
    maxRetries: 1
    minRetryIntervalSeconds: 120
  position:
    x: "200"
    y: "100"

字段说明

字段	类型	必填	说明
`type`	String	是	固定为 `execute_pipeline`。
`pipelineName`	String	是	目标 Pipeline 的 `metadata.name`。
`pipelineId`	Long	是	目标 Pipeline ID（提供时优先使用）。
`pipelineParameterValues`	Map	否	向目标 Pipeline 传递的参数。
`waitingSuccess`	Boolean	否	是否等待子 Pipeline 执行完成（默认 `false`）。
`bizDate`	String	是	子 Pipeline 实例的业务日期。

waitingSuccess 行为

waitingSuccess	行为	适用场景
`true`	阻塞等待子 Pipeline 运行完成后再继续。子 Pipeline 成功则当前 Activity 成功，子 Pipeline 失败则当前 Activity 失败。	下游节点依赖子 Pipeline 的执行结果。
`false`（默认）	仅触发子 Pipeline 后立即返回成功，不等待子 Pipeline 执行完成。	仅需触发执行，不关心结果（如异步通知）。

waitingSuccess true 的执行流程：

waitingSuccess false 的执行流程：

参数传递

通过 pipelineParameterValues 向子 Pipeline 传入参数，覆盖其默认参数值：

pipelineParameterValues:
  biz_date: "${date}"
  check_level: "strict"
  target_table: "dw.order_summary"

说明	详情
覆盖规则	传入的参数会覆盖子 Pipeline 中同名参数的默认值。
未传入的参数	使用子 Pipeline 中定义的默认值。
值来源	支持固定值、内置变量（`${date}`）和父 Pipeline 参数引用。

获取子 Pipeline 返回值

当子 Pipeline 通过 Set Variable 设置了 PIPELINE_OUTPUT 返回值时，父 Pipeline 的后续节点可以引用这些值：

# 子 Pipeline（data_quality_check）中的 Set Variable 节点设置了 output 变量：
# - total_records
# - status

# 父 Pipeline 中后续节点引用子 Pipeline 的返回值
- name: check_result
  type: if_condition
  condition:
    left: "{{pipeline(data_quality_check).output.status}}"
    op: EQUAL_TO
    right: "pass"
  dependsOn:
    activities:
      - run_quality_check

注意

如果后续节点需要读取子 Pipeline 的输出，建议必须设置 waitingSuccess: true，确保当前节点在子 Pipeline 完成后再放行下游。实际读取语法为 {{pipeline(name).output.xxx}}，由参数解析阶段从调度侧查询输出。

典型场景

场景一：ETL + 数据质量检查

activities:
  - name: run_etl
    type: sql
    source: WORKSPACE
    path: /Workspace/Users/zhang3/sql/etl.sql
    engineType: emr_serverless_spark
    engineQueue: default
    position:
      x: "200"
      y: "100"

  - name: run_quality_check
    type: execute_pipeline
    pipelineName: data_quality_check
    pipelineParameterValues:
      target_table: "dw.order_summary"
      biz_date: "${date}"
    waitingSuccess: true
    bizDate: "${date}"
    dependsOn:
      activities:
        - run_etl
    position:
      x: "400"
      y: "100"

场景二：For Each + Execute Pipeline

- name: process_all_regions
  type: for_each
  items: '["cn-beijing", "cn-shanghai", "us-east"]'
  concurrency: 3
  activities:
    - name: run_region_etl
      type: execute_pipeline
      pipelineName: region_etl_pipeline
      pipelineParameterValues:
        region: "{{items.each}}"
        biz_date: "${date}"
      waitingSuccess: true
      bizDate: "${date}"
      position:
        x: "100"
        y: "50"

使用建议

建议	说明
需要结果时设 waitingSuccess: true	下游依赖子 Pipeline 结果时务必设为 true。
子 Pipeline 保持独立可运行	子 Pipeline 应能独立调试和运行，不应强依赖父 Pipeline 上下文。
避免循环调用	不要出现 A → B → A 的 Pipeline 调用链，会导致死锁。
传递关键参数	将 bizDate 和业务参数显式传入，不依赖子 Pipeline 的默认值。

最近更新时间：2026.07.01 00:44:44

这个页面对您有帮助吗？

有用

无用

大数据研发治理套件

概述 #

配置示例 #

字段说明 #

waitingSuccess 行为 #

参数传递 #

获取子 Pipeline 返回值 #

典型场景 #

场景一：ETL + 数据质量检查 #

场景二：For Each + Execute Pipeline #

使用建议 #

概述