You need to enable JavaScript to run this app.
文档中心
大数据研发治理套件

大数据研发治理套件

请输入
  • 文档首页
    • 大数据研发治理套件用户指南数据开发 - New IDE参数、变量与表达式参数试用示例
复制全文
下载 pdf
我的收藏
参数、变量与表达式
参数试用示例
复制全文
下载 pdf
我的收藏
参数试用示例

本文通过典型场景演示参数系统的设计模式，每个示例包含完整的 Pipeline YAML 配置和脚本代码，可直接参考使用。

示例一：参数查找优先级

以下示例展示当同名参数出现在不同层级时，系统如何按优先级解析。
配置：

# Pipeline 定义
spec:
  parameters:
    - name: env
      default: "staging"
    - name: biz_date
      default: "${date}"

  activities:
    - name: my_task
      type: sql
      source: WORKSPACE
      path: /Workspace/Users/zhang3/sql/query.sql
      engineType: emr_serverless_spark
      engineQueue: default
      parameterValues:
        env: "dev"

脚本中引用的解析结果：

脚本中的引用

查找结果

来源

{{env}}

"dev"

Activity parameterValues（最高优先级）

{{biz_date}}

"${date}" 的具体值

Pipeline parameters（Activity 未定义，上溯）

{{db_host}}

"10.0.1.100"

项目参数（Pipeline 也未定义，继续上溯）

{{undefined_param}}

报错

所有层级均未找到

示例二：父子 Pipeline 参数传递

需求： 主 Pipeline 调用数据加工和质量检查两个子 Pipeline，需要向子 Pipeline 传递参数并根据执行结果决定后续流程。
主 Pipeline 配置：

# main_pipeline.pipeline.yml
apiVersion: newide.studio.dataleap.volc/v1
kind: Pipeline
metadata:
  name: main_daily_pipeline
spec:
  displayName: "主调度管道"
  runAs: User/100001
  schedulingResourceGroupName: default_scheduler

  parameters:
    - name: biz_date
      default: "${date}"
    - name: check_level
      default: "standard"

  variables:
    - name: quality_passed
      default: "false"

  activities:
    # 步骤 1：调用数据加工子 Pipeline
    - name: run_etl
      type: execute_pipeline
      pipelineName: data_etl_pipeline
      pipelineParameterValues:
        biz_date: "${date}"
        run_mode: "incremental"
      waitingSuccess: true
      bizDate: "${date}"

    # 步骤 2：调用数据质量检查子 Pipeline
    - name: run_quality_check
      type: execute_pipeline
      pipelineName: quality_check_pipeline
      pipelineParameterValues:
        biz_date: "${date}"
        check_level: "{{pipeline.params.check_level}}"
        source_table: "{{activity(run_etl).output.output_table}}"
      waitingSuccess: true
      bizDate: "${date}"
      dependsOn:
        activities:
          - name: run_etl

    # 步骤 3：根据质量检查结果决定后续流程
    - name: check_result
      type: if_condition
      condition:
        left: "{{activity(run_quality_check).output.check_status}}"
        op: EQUAL_TO
        right: "passed"
      dependsOn:
        activities:
          - name: run_quality_check

    # 步骤 4a：质量通过 → 发布数据
    - name: publish_data
      type: sql
      source: WORKSPACE
      path: /Workspace/Users/zhang3/sql/publish.sql
      engineType: emr_serverless_spark
      engineQueue: default
      parameterValues:
        biz_date: "${date}"
      dependsOn:
        activities:
          - name: check_result
            outcome: true

    # 步骤 4b：质量不通过 → 发送告警
    - name: send_alert
      type: notebook
      source: WORKSPACE
      path: /Workspace/Users/zhang3/scripts/alert.ipynb
      computingResourceGroupName: default_python_group
      parameterValues:
        biz_date: "${date}"
        error_detail: "{{activity(run_quality_check).output.error_message}}"
      dependsOn:
        activities:
          - name: check_result
            outcome: false

  trigger:
    type: scheduled
    cronExpression: "0 2 * * *"
    frequency: daily
    priority: d3

子 Pipeline（设置输出值）：

# quality_check_pipeline.pipeline.yml
apiVersion: newide.studio.dataleap.volc/v1
kind: Pipeline
metadata:
  name: quality_check_pipeline
spec:
  displayName: "数据质量检查"
  runAs: User/100001
  schedulingResourceGroupName: default_scheduler

  parameters:
    - name: biz_date
      default: "${date}"
    - name: check_level
      default: "standard"
    - name: source_table
      default: ""

  variables:
    - name: check_result
      default: "unknown"
    - name: error_msg
      default: ""

  activities:
    - name: run_checks
      type: sql
      source: WORKSPACE
      path: /Workspace/Users/zhang3/sql/quality_checks.sql
      engineType: emr_serverless_spark
      engineQueue: default
      parameterValues:
        biz_date: "${date}"
        table_name: "{{pipeline.params.source_table}}"

    # 将结果设为 Pipeline 输出
    - name: set_output
      type: set_variable
      variableType: PIPELINE_OUTPUT
      outputVariables:
        - name: check_status
          default: "{{pipeline.vars.check_result}}"
        - name: error_message
          default: "{{pipeline.vars.error_msg}}"
      dependsOn:
        activities:
          - name: run_checks

  trigger:
    type: manual

关键设计点：

  • 子 Pipeline 的 trigger.type 设为 manual，仅通过父 Pipeline 的 execute_pipeline 触发。
  • 父 Pipeline 通过 pipelineParameterValues 向子 Pipeline 传参。
  • 子 Pipeline 通过 set_variable（PIPELINE_OUTPUT）设置返回值。
  • 父 Pipeline 通过 {{activity(<execute_pipeline_name>).output.<field>}} 读取返回值。
  • waitingSuccess: true 是获取子 Pipeline 输出的前提条件。

示例三：精确引用在配置各处的使用

以下示例汇总展示精确引用语法在不同配置字段中的使用方式：

apiVersion: newide.studio.dataleap.volc/v1
kind: Pipeline
metadata:
  name: reference_demo
spec:
  displayName: "引用语法综合示例"
  runAs: User/100001
  schedulingResourceGroupName: default_scheduler

  parameters:
    - name: biz_date
      default: "${date}"                                          # 系统变量
    - name: upstream_result
      default: "{{pipeline(upstream_etl).output.status}}"         # 上游 Pipeline 输出

  variables:
    - name: processed_count
      default: "0"

  activities:
    # sql — parameterValues 中引用项目参数和 Pipeline 参数
    - name: extract
      type: sql
      source: WORKSPACE
      path: /Workspace/Users/{{project.code_owner}}/sql/extract.sql  # path 中引用项目参数
      engineType: emr_serverless_spark
      engineQueue: default
      parameterValues:
        source_db: "{{project.source_database}}"       # 项目参数
        biz_date: "{{pipeline.params.biz_date}}"              # Pipeline 参数

    # python — arguments 中引用 Pipeline 参数
    - name: transform
      type: python
      source: WORKSPACE
      path: /Workspace/Users/zhang3/scripts/transform.py
      computingResourceGroupName: default_python_group
      arguments:
        - "--biz_date"
        - "{{pipeline.params.biz_date}}"                      # arguments 中引用
      parameterValues:
        upstream_status: "{{pipeline.params.upstream_result}}" # parameterValues 中引用
      dependsOn:
        activities:
          - name: extract

    # if_condition — condition 中引用变量
    - name: check_count
      type: if_condition
      condition:
        left: "{{pipeline.vars.processed_count}}"                     # 变量引用
        op: GREATER_THAN
        right: "0"
      dependsOn:
        activities:
          - name: transform

    # for_each — items 中引用上游 Activity 输出
    - name: process_items
      type: for_each
      items: "{{activity(extract).output.rows}}"                 # Activity 输出引用
      concurrency: 3
      activities:
        - name: item_task
          type: sql
          source: WORKSPACE
          path: /Workspace/Users/zhang3/sql/item.sql
          engineType: emr_serverless_spark
          engineQueue: default
          parameterValues:
            item_id: "{{items.each}}"                             # 迭代元素引用
      dependsOn:
        activities:
          - name: check_count
            outcome: true

    # execute_pipeline — pipelineParameterValues 中引用
    - name: call_sub
      type: execute_pipeline
      pipelineName: sub_pipeline
      pipelineParameterValues:
        biz_date: "${date}"                                       # 系统变量
        source_table: "{{project.source_database}}"    # 项目参数
      waitingSuccess: true
      bizDate: "${date}"
      dependsOn:
        activities:
          - name: check_count
            outcome: true

    # set_variable — rawValue / calculateValue / outputVariables 中引用
    - name: update_count
      type: set_variable
      variableType: PIPELINE_INNER
      variableName: processed_count
      value:
        calculateValue:
          left: "{{pipeline.vars.processed_count}}"                   # 变量引用
          op: PLUS
          right: "{{activity(extract).output.row_count}}"        # Activity 输出引用
      dependsOn:
        activities:
          - name: process_items

    - name: set_output
      type: set_variable
      variableType: PIPELINE_OUTPUT
      outputVariables:
        - name: total
          default: "{{pipeline.vars.processed_count}}"                # 变量引用
      dependsOn:
        activities:
          - name: update_count

  dependsOn:
    pipelines:
      - name: upstream_etl
        offsetsType: set
        offsets:
          - "0"

  trigger:
    type: scheduled
    cronExpression: "0 2 * * *"
    frequency: daily

参数设计最佳实践

命名规范

规范

说明

正例

反例

使用 snake_case

小写字母 + 下划线

biz_date

bizDate

语义清晰

名称应体现参数含义

source_database

src_db

添加前缀区分来源

避免不同类型参数混淆

target_region

region

统一日期命名

项目内保持一致

biz_date

dt / date / day

分层原则

原则

说明

作用域最小化

仅脚本内部使用的参数在 Activity 级传值，不要提升到 Pipeline 级。

环境配置上提

环境相关的值（数据库、路径）定义为项目参数。

业务参数中间化

运行时传入的业务参数（日期、模式）定义为 Pipeline 参数。

状态传递用变量

Activity 间传递动态状态使用 Pipeline 变量。

跨 Pipeline 用输出

需要传给下游 Pipeline 的值使用 PIPELINE_OUTPUT。

默认值设计

实践

说明

所有参数必须提供默认值

确保手动触发时可直接运行，无需额外指定。

日期参数使用系统变量

default: "${date}" 自动取当前业务日期。

环境参数引用项目参数

default: "{{project.xxx}}" 自动适配环境。

避免硬编码

可变的值不应写死在脚本或配置中。

安全实践

实践

说明

敏感信息用 Secret 类型

密码、Token、密钥必须使用 secret 类型项目参数。

不在脚本中明文写密码

通过参数传入，保持代码可安全提交到 Git。

权限最小化

Secret 参数仅 Admin 角色可创建和修改。

最近更新时间：2026.07.01 00:44:44
这个页面对您有帮助吗？
有用
有用
无用
无用