本文介绍 Execute Pipeline 类型的 Activity 配置，用于在当前 Pipeline 中触发执行另一个已发布的 Pipeline。

概述 #

Execute Pipeline Activity 实现子 Pipeline 调用，将复杂的工作流拆分为多个可复用的 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
    minRetryIntervalMillis: 120000
  position:
    x: "200"
    y: "100"

字段说明 #

waitingSuccess 行为 #

waitingSuccess: true 的执行流程：

Execute Pipeline 节点开始
  → 触发子 Pipeline 生成实例
    → 子 Pipeline 开始执行
      → 等待子 Pipeline 执行完成...
        → 子 Pipeline SUCCEEDED → 当前 Activity SUCCEEDED
        → 子 Pipeline FAILED → 当前 Activity FAILED

waitingSuccess: false 的执行流程：

Execute Pipeline 节点开始
  → 触发子 Pipeline 生成实例
    → 立即返回 → 当前 Activity SUCCEEDED
    （子 Pipeline 在后台异步执行）

参数传递 #

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

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

跨项目调用 #

跨项目调用需同时提供 projectId 和 pipelineName：

- name: call_external_pipeline
  type: execute_pipeline
  pipelineName: dim_table_etl
  projectId: 789                       # 目标项目 ID
  pipelineParameterValues:
    biz_date: "${date}"
  waitingSuccess: true
  bizDate: "${date}"
  position:
    x: "200"
    y: "100"

跨项目调用要求：

获取子 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: "{{activities.run_quality_check.output.status}}"
    op: EQUAL_TO
    right: "pass"
  dependsOn:
    activities:
      - run_quality_check

典型场景 #

场景一：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 #