如何使用Azure ML SDK v2的PipelineJob类设置管道最终输出

使用Azure Machine Learning SDK v2构建机器学习管道，已成功创建多个command步骤并加入管道，管道运行正常，但无法将第三步的输出设置为管道最终输出。已知可通过dsl @pipeline装饰器实现，但希望直接通过PipelineJob类完成该操作，附上相关代码及尝试过的两种配置方式，寻求解决方案。

相关代码 #

# Tags
ENVIRONMENT = "env@latest"
TAGS = {"Created_by":"user@company.com"}

# Step 1
step_1 = command(
    name = "step_1_name",
    display_name="Step1_displayname",
    description="One time task for step 1",
    environment=ENVIRONMENT,
    tags=TAGS,
    code = "../code_path",
    inputs = {
        'input_1':Input(
            type='string',
            default="default_value_1",
            description="input 1 description"
            ),
        'input_2':input_data,
        },
    outputs= {
        'output_1': Output(
            type=AssetTypes.URI_FILE,
            mode=InputOutputModes.RW_MOUNT)
        },
    command="""python ./src/script1.py \
            --input_1 ${{inputs.input_1}} \
            --input_2 ${{inputs.input_2}} \
            --output_1 ${{outputs.output_1}}"""
)

# Step 2
step_2 = command(
    name = "step_2_name",
    display_name="Step2_displayname",
    description = "One time task for step 2",
    environment=ENVIRONMENT,
    tags=TAGS,
    code = "../code_path",
    inputs = {
        'input_1':Input(
            type='string',
            default="default_value_2",
            description="input 1 description"
            ),
        'input_2': Input(
            path=step_1.outputs.output_1,
            type= AssetTypes.URI_FILE,
            mode=InputOutputModes.RO_MOUNT
            ),
    },
    outputs= {
        'output_1':Output(type=AssetTypes.URI_FILE, mode=InputOutputModes.RW_MOUNT),
        'output_2':Output(type=AssetTypes.URI_FILE, mode=InputOutputModes.RW_MOUNT),
        'output_3':Output(type=AssetTypes.URI_FILE, mode=InputOutputModes.RW_MOUNT),
        'output_4':Output(type=AssetTypes.URI_FILE, mode=InputOutputModes.RW_MOUNT)
    },
    command="""python ./src/script2.py \
            --input_1 ${{inputs.input_1}} \
            --input_2 ${{inputs.input_2}} \
            --output_1 ${{outputs.output_1}} \
            --output_2 ${{outputs.output_2}} \
            --output_3 ${{outputs.output_3}} \
            --output_4 ${{outputs.output_4}}"""
)

# Step 3
step_3 = command(
    name = "step_3_name",
    display_name="Step3_displayname",
    description="One time task for step 3",
    environment=ENVIRONMENT,
    tags=TAGS,
    code = "../code_path",
    inputs={
        'input_1':Input(type='string', default="default_value_3", description="input 1 description"),
        'input_2': Input(path=step_2.outputs.output_1, type= AssetTypes.URI_FILE, mode=InputOutputModes.RO_MOUNT),
    },
    outputs= {'output_1':Output(type=AssetTypes.URI_FILE, mode=InputOutputModes.RW_MOUNT)},
    command="""python ./src/script3.py \
            --input_1 ${{inputs.input_1}} \
            --input_2 ${{inputs.input_2}} \
            --output_1 ${{outputs.output_1}}"""
)

# Creating Pipeline
pipe_job = PipelineJob(
    display_name="pipeline_display_name",
    description="pipeline description",
    jobs={
        "step_1": step_1,
        "step_2": step_2,
        "step_3": step_3
    },
    outputs={
        "output_1": Output(type=AssetTypes.URI_FILE, path=step_3.outputs.output_1),
    },
    settings=PipelineJobSettings(force_rerun=True),
    compute="compute_target",
    identity=UserIdentityConfiguration()
)

# Submit the pipeline job
pipeline_job_response = ml_client.jobs.create_or_update(pipe_job)
print("Pipeline job submitted:", pipeline_job_response.name)
pipeline_job_result = ml_client.jobs.stream(pipeline_job_response.name)

已尝试的配置 #

• 配置1：outputs={"pipe_path": "${jobs.step_3.outputs.output_1}")}
• 配置2：outputs={"pipe_path": step_3.outputs.pipe_path}

解决方案 #

方案1：直接引用步骤输出对象 #

在PipelineJob的outputs中直接传递step_3的输出对象，SDK会自动处理关联：

pipe_job = PipelineJob(
    display_name="pipeline_display_name",
    description="pipeline description",
    jobs={
        "step_1": step_1,
        "step_2": step_2,
        "step_3": step_3
    },
    outputs={
        "final_output": step_3.outputs.output_1,
    },
    settings=PipelineJobSettings(force_rerun=True),
    compute="compute_target",
    identity=UserIdentityConfiguration()
)

方案2：使用正确的管道表达式语法 #

使用双花括号${{...}}的管道表达式，注意语法格式正确：

pipe_job = PipelineJob(
    display_name="pipeline_display_name",
    description="pipeline description",
    jobs={
        "step_1": step_1,
        "step_2": step_2,
        "step_3": step_3
    },
    outputs={
        "final_output": "${{jobs.step_3.outputs.output_1}}",
    },
    settings=PipelineJobSettings(force_rerun=True),
    compute="compute_target",
    identity=UserIdentityConfiguration()
)

问题说明 #

• 之前尝试的配置1存在语法错误，多了一个闭合括号，且表达式应使用双花括号
• 配置2引用了不存在的输出字段pipe_path，正确字段应为step_3.outputs.output_1
• 无需重新创建Output实例，直接关联步骤输出即可完成管道最终输出的设置

内容的提问来源于stack exchange，提问作者Marvin