如何在Azure ML Service中使用参数化Dataset/DataPath输入进行批量推理并向ParallelRunStep传递参数化Dataset输入

我最近刚碰到这个问题，折腾了一阵终于找到可行的解决方案，分享给你：

问题根源 #

直接把用PipelineParameter创建的DatasetConsumptionConfig同时放到ParallelRunStep的arguments和inputs列表里会报错——这是因为ParallelRunStep对输入的处理逻辑和普通PythonScriptStep不同，它不支持将DatasetConsumptionConfig类型的对象传入arguments参数，仅允许在inputs中传递这类配置。

完整解决方案步骤 #

1. 创建参数化的Dataset配置

首先定义带默认值的PipelineParameter，再关联到DatasetConsumptionConfig：

# 1. 创建默认数据集（可选，也可以不设置默认值）
default_tabular_ds = Dataset.Tabular.from_delimited_files('https://dprepdata.blob.core.windows.net/demo/Titanic.csv')
# 2. 创建PipelineParameter，用于动态替换数据集
tabular_ds_param = PipelineParameter(name="tabular_ds_param", default_value=default_tabular_ds)
# 3. 创建DatasetConsumptionConfig，这里的名称（第一个参数）要和后续脚本中引用的一致
tabular_ds_consumption = DatasetConsumptionConfig("my_param_input", tabular_ds_param)

2. 配置ParallelRunConfig和ParallelRunStep

重点：仅将DatasetConsumptionConfig传入ParallelRunStep的inputs列表，不要放到arguments里：

from azureml.contrib.pipeline.steps import ParallelRunConfig, ParallelRunStep

# 定义ParallelRunConfig
parallel_run_config = ParallelRunConfig(
    source_directory="./batch_scripts",
    entry_script="batch_processor.py",  # 你的批量处理脚本
    mini_batch_size="10",  # 根据你的数据调整
    error_threshold=10,
    output_action="append_row",
    compute_target=your_compute_target,  # 提前创建好的计算目标
    node_count=2,
    environment=your_custom_env  # 提前定义好的运行环境
)

# 创建ParallelRunStep
parallel_process_step = ParallelRunStep(
    name="param-driven-parallel-step",
    inputs=[tabular_ds_consumption],
    output=your_output_dataset,  # 提前定义的输出Dataset
    parallel_run_config=parallel_run_config,
    allow_reuse=False
)

3. 在处理脚本中获取参数化输入

在你的batch_processor.py脚本里，可以通过环境变量或者Run上下文获取挂载的输入路径：

# batch_processor.py
import os
from azureml.core import Run

run = Run.get_context()

# 方式1：通过环境变量获取（环境变量名称为AZUREML_INPUT_<输入名称>，全大写）
input_data_path = os.environ.get("AZUREML_INPUT_MY_PARAM_INPUT")

# 方式2：通过Run上下文获取输入
input_data = run.input_datasets["my_param_input"]
input_data_path = input_data.to_path()[0]

# 接下来处理路径下的文件
for root, _, files in os.walk(input_data_path):
    for file in files:
        file_full_path = os.path.join(root, file)
        # 你的批量处理逻辑，比如读取文件、处理数据等

4. 提交带新参数的管道运行

当需要替换数据集时，创建新的Dataset实例并通过pipeline_parameters传入：

# 创建新的目标数据集
new_iris_ds = Dataset.Tabular.from_delimited_files("your_new_data_storage_link")

# 提交管道，指定参数替换默认值
pipeline_run = experiment.submit(your_pipeline, pipeline_parameters={"tabular_ds_param": new_iris_ds})

关键注意点 #

• 不要将DatasetConsumptionConfig传入ParallelRunStep的arguments参数，这是触发报错的主要原因；
• DatasetConsumptionConfig的名称要和脚本中引用的输入名称完全一致（环境变量会自动转为大写，加上AZUREML_INPUT_前缀）；
• 确保你的计算环境中已经安装了必要的依赖（比如azureml-core、azureml-dataset-runtime等）。

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