CLI (v1) 管道作业 YAML 架构

项目
2023/07/24

适用于：Azure CLI ml 扩展 v1

备注

本文档中详细介绍的 YAML 语法基于 ML CLI 扩展 v1 版本的 JSON 架构。此语法仅保证适用于 ML CLI v1 扩展。若要了解 ML CLI v2 的语法，请切换到 v2（当前版本）。

重要

本文中的 Azure CLI 命令需要将 azure-cli-ml、v1 或扩展用于 Azure 机器学习。对 v1 扩展的支持将于 2025 年 9 月 30 日结束。在该日期之前，你将能够安装和使用 v1 扩展。

建议在 2025 年 9 月 30 日之前转换为 ml 或 v2 扩展。有关 v2 扩展的详细信息，请参阅 Azure ML CLI 扩展和 Python SDK v2。

在 YAML 中定义机器学习管道。使用适用于 Azure CLI v1 的机器学习扩展时，许多管道相关的命令都要求提供一个用于定义管道的 YAML 文件。

下表列出了在 YAML 中定义用于 CLI v1 的管道时目前支持和不支持的内容：

步骤类型	支持？
PythonScriptStep	是
ParallelRunStep	是
AdlaStep	是
AzureBatchStep	是
DatabricksStep	是
DataTransferStep	是
AutoMLStep	否
HyperDriveStep	否
ModuleStep	是
MPIStep	否
EstimatorStep	否

管道定义

管道定义使用以下对应于 Pipelines 类的键：

YAML 键	说明
`name`	管道的说明。
`parameters`	管道的参数。
`data_reference`	定义在运行中提供数据的方式和位置。
`default_compute`	默认的计算目标，管道中的所有步骤将在其上运行。
`steps`	管道中使用的步骤。

parameters

parameters 节使用以下对应于 PipelineParameter 类的键：

YAML 键	说明
`type`	参数的值类型。有效类型为 `string`、`int`、`float`、`bool` 或 `datapath`。
`default`	默认值。

每个参数已命名。例如，以下 YAML 代码片段定义名为 NumIterationsParameter、DataPathParameter 和 NodeCountParameter 的三个参数：

pipeline:
    name: SamplePipelineFromYaml
    parameters:
        NumIterationsParameter:
            type: int
            default: 40
        DataPathParameter:
            type: datapath
            default:
                datastore: workspaceblobstore
                path_on_datastore: sample2.txt
        NodeCountParameter:
            type: int
            default: 4

数据引用

data_references 节使用以下对应于 DataReference 的键：

YAML 键	说明
`datastore`	要引用的数据存储。
`path_on_datastore`	数据引用在后备存储中的相对路径。

每个数据引用包含在一个键中。例如，以下 YAML 代码片段定义存储在键中的名为 employee_data 的数据引用：

pipeline:
    name: SamplePipelineFromYaml
    parameters:
        PipelineParam1:
            type: int
            default: 3
    data_references:
        employee_data:
            datastore: adftestadla
            path_on_datastore: "adla_sample/sample_input.csv"

步骤

步骤定义计算环境，以及要在环境中运行的文件。若要定义步骤的类型，请使用 type 键：

步骤类型	说明
`AdlaStep`	使用 Azure Data Lake Analytics 运行 U-SQL 脚本。对应于 AdlaStep 类。
`AzureBatchStep`	使用 Azure Batch 运行作业。对应于 AzureBatchStep 类。
`DatabricsStep`	添加 Databricks 笔记本、Python 脚本或 JAR。对应于 DatabricksStep 类。
`DataTransferStep`	在存储选项之间传输数据。对应于 DataTransferStep 类。
`PythonScriptStep`	运行 Python 脚本。对应于 PythonScriptStep 类。
`ParallelRunStep`	运行 Python 脚本，以异步方式并行处理大量数据。对应于 ParallelRunStep 类。

ADLA 步骤

YAML 键	说明
`script_name`	U-SQL 脚本的名称（相对于 `source_directory`）。
`compute`	用于此步骤的 Azure Data Lake 计算目标。
`parameters`	管道的参数。
`inputs`	输入可以是 InputPortBinding、DataReference、PortDataReference、PipelineData、Dataset、DatasetDefinition 或 PipelineDataset。
`outputs`	输出可以是 PipelineData 或 OutputPortBinding。
`source_directory`	包含脚本、程序集等的目录。
`priority`	用于当前作业的优先级值。
`params`	名称/值对的字典。
`degree_of_parallelism`	用于此作业的并行度。
`runtime_version`	Data Lake Analytics 引擎的运行时版本。
`allow_reuse`	确定当使用相同的设置再次运行时，该步骤是否应重用以前的结果。

以下示例包含 ADLA 步骤定义：

pipeline:
    name: SamplePipelineFromYaml
    parameters:
        PipelineParam1:
            type: int
            default: 3
    data_references:
        employee_data:
            datastore: adftestadla
            path_on_datastore: "adla_sample/sample_input.csv"
    default_compute: adlacomp
    steps:
        Step1:
            runconfig: "D:\\Yaml\\default_runconfig.yml"
            parameters:
                NUM_ITERATIONS_2:
                    source: PipelineParam1
                NUM_ITERATIONS_1: 7
            type: "AdlaStep"
            name: "MyAdlaStep"
            script_name: "sample_script.usql"
            source_directory: "D:\\scripts\\Adla"
            inputs:
                employee_data:
                    source: employee_data
            outputs:
                OutputData:
                    destination: Output4
                    datastore: adftestadla
                    bind_mode: mount

Azure Batch 步骤

YAML 键	说明
`compute`	用于此步骤的 Azure Batch 计算目标。
`inputs`	输入可以是 InputPortBinding、DataReference、PortDataReference、PipelineData、Dataset、DatasetDefinition 或 PipelineDataset。
`outputs`	输出可以是 PipelineData 或 OutputPortBinding。
`source_directory`	包含模块二进制文件、可执行文件、程序集等的目录。
`executable`	要作为此作业的一部分运行的命令/可执行文件的名称。
`create_pool`	布尔标志，指示在运行作业之前是否创建池。
`delete_batch_job_after_finish`	布尔标志，指示在完成作业后是否从 Batch 帐户中删除该作业。
`delete_batch_pool_after_finish`	布尔标志，指示在完成作业后是否删除池。
`is_positive_exit_code_failure`	布尔标志，指示在任务退出并返回正代码时作业是否失败。
`vm_image_urn`	如果 `create_pool` 为 `True`，则 VM 将使用 `VirtualMachineConfiguration`。
`pool_id`	要在其中运行作业的池的 ID。
`allow_reuse`	确定当使用相同的设置再次运行时，该步骤是否应重用以前的结果。

以下示例包含 Azure Batch 步骤定义：

pipeline:
    name: SamplePipelineFromYaml
    parameters:
        PipelineParam1:
            type: int
            default: 3
    data_references:
        input:
            datastore: workspaceblobstore
            path_on_datastore: "input.txt"
    default_compute: testbatch
    steps:
        Step1:
            runconfig: "D:\\Yaml\\default_runconfig.yml"
            parameters:
                NUM_ITERATIONS_2:
                    source: PipelineParam1
                NUM_ITERATIONS_1: 7
            type: "AzureBatchStep"
            name: "MyAzureBatchStep"
            pool_id: "MyPoolName"
            create_pool: true
            executable: "azurebatch.cmd"
            source_directory: "D:\\scripts\\AureBatch"
            allow_reuse: false
            inputs:
                input:
                    source: input
            outputs:
                output:
                    destination: output
                    datastore: workspaceblobstore

Databricks 步骤

YAML 键	说明
`compute`	用于此步骤的 Azure Databricks 计算目标。
`inputs`	输入可以是 InputPortBinding、DataReference、PortDataReference、PipelineData、Dataset、DatasetDefinition 或 PipelineDataset。
`outputs`	输出可以是 PipelineData 或 OutputPortBinding。
`run_name`	此运行在 Databricks 中的名称。
`source_directory`	包含脚本和其他文件的目录。
`num_workers`	Databricks 运行群集的辅助角色的静态编号。
`runconfig`	`.runconfig` 文件的路径。此文件是 RunConfiguration 类的 YAML 表示形式。有关此文件的结构的详细信息，请参阅 runconfigschema.json。
`allow_reuse`	确定当使用相同的设置再次运行时，该步骤是否应重用以前的结果。

以下示例包含 Databricks 步骤：

pipeline:
    name: SamplePipelineFromYaml
    parameters:
        PipelineParam1:
            type: int
            default: 3
    data_references:
        adls_test_data:
            datastore: adftestadla
            path_on_datastore: "testdata"
        blob_test_data:
            datastore: workspaceblobstore
            path_on_datastore: "dbtest"
    default_compute: mydatabricks
    steps:
        Step1:
            runconfig: "D:\\Yaml\\default_runconfig.yml"
            parameters:
                NUM_ITERATIONS_2:
                    source: PipelineParam1
                NUM_ITERATIONS_1: 7
            type: "DatabricksStep"
            name: "MyDatabrickStep"
            run_name: "DatabricksRun"
            python_script_name: "train-db-local.py"
            source_directory: "D:\\scripts\\Databricks"
            num_workers: 1
            allow_reuse: true
            inputs:
                blob_test_data:
                    source: blob_test_data
            outputs:
                OutputData:
                    destination: Output4
                    datastore: workspaceblobstore
                    bind_mode: mount

数据传输步骤

YAML 键	说明
`compute`	用于此步骤的 Azure 数据工厂计算目标。
`source_data_reference`	充当数据传输操作的源的输入连接。支持的值为 InputPortBinding、DataReference、PortDataReference、PipelineData、Dataset、DatasetDefinition 或 PipelineDataset。
`destination_data_reference`	充当数据传输操作的目标的输入连接。支持的值为 PipelineData 和 OutputPortBinding。
`allow_reuse`	确定当使用相同的设置再次运行时，该步骤是否应重用以前的结果。

以下示例包含数据传输步骤：

pipeline:
    name: SamplePipelineFromYaml
    parameters:
        PipelineParam1:
            type: int
            default: 3
    data_references:
        adls_test_data:
            datastore: adftestadla
            path_on_datastore: "testdata"
        blob_test_data:
            datastore: workspaceblobstore
            path_on_datastore: "testdata"
    default_compute: adftest
    steps:
        Step1:
            runconfig: "D:\\Yaml\\default_runconfig.yml"
            parameters:
                NUM_ITERATIONS_2:
                    source: PipelineParam1
                NUM_ITERATIONS_1: 7
            type: "DataTransferStep"
            name: "MyDataTransferStep"
            adla_compute_name: adftest
            source_data_reference:
                adls_test_data:
                    source: adls_test_data
            destination_data_reference:
                blob_test_data:
                    source: blob_test_data

Python 脚本步骤

YAML 键	说明
`inputs`	输入可以是 InputPortBinding、DataReference、PortDataReference、PipelineData、Dataset、DatasetDefinition 或 PipelineDataset。
`outputs`	输出可以是 PipelineData 或 OutputPortBinding。
`script_name`	Python 脚本的名称（相对于 `source_directory`）。
`source_directory`	包含脚本、Conda 环境等的目录。
`runconfig`	`.runconfig` 文件的路径。此文件是 RunConfiguration 类的 YAML 表示形式。有关此文件的结构的详细信息，请参阅 runconfig.json。
`allow_reuse`	确定当使用相同的设置再次运行时，该步骤是否应重用以前的结果。

以下示例包含 Python 脚本步骤：

pipeline:
    name: SamplePipelineFromYaml
    parameters:
        PipelineParam1:
            type: int
            default: 3
    data_references:
        DataReference1:
            datastore: workspaceblobstore
            path_on_datastore: testfolder/sample.txt
    default_compute: cpu-cluster
    steps:
        Step1:
            runconfig: "D:\\Yaml\\default_runconfig.yml"
            parameters:
                NUM_ITERATIONS_2:
                    source: PipelineParam1
                NUM_ITERATIONS_1: 7
            type: "PythonScriptStep"
            name: "MyPythonScriptStep"
            script_name: "train.py"
            allow_reuse: True
            source_directory: "D:\\scripts\\PythonScript"
            inputs:
                InputData:
                    source: DataReference1
            outputs:
                OutputData:
                    destination: Output4
                    datastore: workspaceblobstore
                    bind_mode: mount

并行运行步骤

YAML 键	说明
`inputs`	输入可以是 Dataset、DatasetDefinition 或 PipelineDataset。
`outputs`	输出可以是 PipelineData 或 OutputPortBinding。
`script_name`	Python 脚本的名称（相对于 `source_directory`）。
`source_directory`	包含脚本、Conda 环境等的目录。
`parallel_run_config`	`parallel_run_config.yml` 文件的路径。此文件是 ParallelRunConfig 类的 YAML 表示形式。
`allow_reuse`	确定当使用相同的设置再次运行时，该步骤是否应重用以前的结果。

以下示例包含并行运行步骤：

pipeline:
    description: SamplePipelineFromYaml
    default_compute: cpu-cluster
    data_references:
        MyMinistInput:
            dataset_name: mnist_sample_data
    parameters:
        PipelineParamTimeout:
            type: int
            default: 600
    steps:        
        Step1:
            parallel_run_config: "yaml/parallel_run_config.yml"
            type: "ParallelRunStep"
            name: "parallel-run-step-1"
            allow_reuse: True
            arguments:
            - "--progress_update_timeout"
            - parameter:timeout_parameter
            - "--side_input"
            - side_input:SideInputData
            parameters:
                timeout_parameter:
                    source: PipelineParamTimeout
            inputs:
                InputData:
                    source: MyMinistInput
            side_inputs:
                SideInputData:
                    source: Output4
                    bind_mode: mount
            outputs:
                OutputDataStep2:
                    destination: Output5
                    datastore: workspaceblobstore
                    bind_mode: mount

包含多个步骤的管道

YAML 键	说明
`steps`	具有一个或多个 PipelineStep 定义的序列。请注意，步骤 `outputs` 的 `destination` 键将成为下一步 `inputs` 的 `source` 键。

pipeline:
    name: SamplePipelineFromYAML
    description: Sample multistep YAML pipeline
    data_references:
        TitanicDS:
            dataset_name: 'titanic_ds'
            bind_mode: download
    default_compute: cpu-cluster
    steps:
        Dataprep:
            type: "PythonScriptStep"
            name: "DataPrep Step"
            compute: cpu-cluster
            runconfig: ".\\default_runconfig.yml"
            script_name: "prep.py"
            arguments:
            - '--train_path'
            - output:train_path
            - '--test_path'
            - output:test_path
            allow_reuse: True
            inputs:
                titanic_ds:
                    source: TitanicDS
                    bind_mode: download
            outputs:
                train_path:
                    destination: train_csv
                    datastore: workspaceblobstore
                test_path:
                    destination: test_csv
        Training:
            type: "PythonScriptStep"
            name: "Training Step"
            compute: cpu-cluster
            runconfig: ".\\default_runconfig.yml"
            script_name: "train.py"
            arguments:
            - "--train_path"
            - input:train_path
            - "--test_path"
            - input:test_path
            inputs:
                train_path:
                    source: train_csv
                    bind_mode: download
                test_path:
                    source: test_csv
                    bind_mode: download

计划

为管道定义计划时，该计划可以由数据存储触发，或根据某个时间间隔重复运行。下面是用于定义计划的键：

YAML 键	说明
`description`	计划的说明。
`recurrence`	包含重复设置（如果计划是重复性的）。
`pipeline_parameters`	管道所需的任何参数。
`wait_for_provisioning`	是否等待计划预配完成。
`wait_timeout`	超时之前等待的秒数。
`datastore_name`	要在其中监视已修改/已添加的 Blob 的数据存储。
`polling_interval`	轮询已修改/已添加的 Blob 的间隔时间（分钟）。默认值：5 分钟。仅支持数据存储计划。
`data_path_parameter_name`	要使用更改的 Blob 路径设置的数据路径管道参数的名称。仅支持数据存储计划。
`continue_on_step_failure`	当某个步骤失败时，是否继续执行提交的 PipelineRun 中的其他步骤。如果提供此键，将替代管道的 `continue_on_step_failure` 设置。
`path_on_datastore`	可选。要在其中监视已修改/已添加的 Blob 的数据存储上的路径。该路径位于数据存储的容器下，因此，计划监视的实际路径是 container/`path_on_datastore`。如果没有此路径，将监视数据存储容器。不会监视在 `path_on_datastore` 的子文件夹中进行的添加/修改。仅支持数据存储计划。

以下示例包含数据存储触发的计划的定义：

Schedule: 
      description: "Test create with datastore" 
      recurrence: ~ 
      pipeline_parameters: {} 
      wait_for_provisioning: True 
      wait_timeout: 3600 
      datastore_name: "workspaceblobstore" 
      polling_interval: 5 
      data_path_parameter_name: "input_data" 
      continue_on_step_failure: None 
      path_on_datastore: "file/path"

定义重复性计划时，请在 recurrence 下使用以下键：

YAML 键	说明
`frequency`	计划的重复频率。有效值为 `"Minute"`、`"Hour"`、`"Day"`、`"Week"` 或 `"Month"`。
`interval`	计划的激发频率。整数值为再次激发计划之前要等待的时间单位数。
`start_time`	计划的开始时间。该值的字符串格式为 `YYYY-MM-DDThh:mm:ss`。如果未提供开始时间，则第一个工作负荷会即时运行，以后的工作负荷将按计划运行。如果开始时间是过去的时间，则第一个工作负荷将在下一个计算的运行时间运行。
`time_zone`	开始时间所用的时区。如果未提供时区，将使用 UTC。
`hours`	如果 `frequency` 是 `"Day"` 或 `"Week"`，可以指定从 0 到 23 的一个或多个整数（用逗号分隔），表示一天中应运行管道的小时时间。只能使用 `time_of_day` 或 `hours` 和 `minutes`。
`minutes`	如果 `frequency` 是 `"Day"` 或 `"Week"`，可以指定从 0 到 59 的一个或多个整数（用逗号分隔），表示指定的小时中应运行管道的分钟时间。只能使用 `time_of_day` 或 `hours` 和 `minutes`。
`time_of_day`	如果 `frequency` 是 `"Day"` 或 `"Week"`，可以指定一天中要运行计划的时间。该值的字符串格式为 `hh:mm`。只能使用 `time_of_day` 或 `hours` 和 `minutes`。
`week_days`	如果 `frequency` 是 `"Week"`，可以指定应运行计划的一个或多个星期日期（用逗号分隔）。有效值为 `"Monday"`、`"Tuesday"`、`"Wednesday"`、`"Thursday"`、`"Friday"`、`"Saturday"` 和 `"Sunday"`。

以下示例包含重复性计划的定义：

Schedule: 
    description: "Test create with recurrence" 
    recurrence: 
        frequency: Week # Can be "Minute", "Hour", "Day", "Week", or "Month". 
        interval: 1 # how often fires 
        start_time: 2019-06-07T10:50:00 
        time_zone: UTC 
        hours: 
        - 1 
        minutes: 
        - 0 
        time_of_day: null 
        week_days: 
        - Friday 
    pipeline_parameters: 
        'a': 1 
    wait_for_provisioning: True 
    wait_timeout: 3600 
    datastore_name: ~ 
    polling_interval: ~ 
    data_path_parameter_name: ~ 
    continue_on_step_failure: None 
    path_on_datastore: ~

后续步骤

了解如何使用 Azure 机器学习的 CLI 扩展。

未来属于你

通过

管道定义

parameters

数据引用

步骤

ADLA 步骤

Azure Batch 步骤

Databricks 步骤

数据传输步骤

Python 脚本步骤

并行运行步骤

包含多个步骤的管道

计划

后续步骤

通过

CLI (v1) 管道作业 YAML 架构

管道定义

parameters

数据引用

步骤

ADLA 步骤

Azure Batch 步骤

Databricks 步骤

数据传输步骤

Python 脚本步骤

并行运行步骤

包含多个步骤的管道

计划

后续步骤

其他资源