如何解决SageMaker Pipeline多数据输入预处理的报错问题？

问题概述 #

运行带多个数据输入的SageMaker预处理管道时，单输入场景可正常执行，但添加第二个需合并的数据输入后出现两类错误：

• FailureReason: 'ClientError: AlgorithmError: , exit code: 1'
• FailureReason: 'ClientError: Failed to invoke sagemaker:CreateProcessingJob. Error Details: null'

备注：单独使用任一数据输入均可成功运行，说明管道具备数据读取能力，问题出在多参数/输入的配置逻辑上。

当前代码配置 #

参数定义 #

from sagemaker.workflow.parameters import (
ParameterInteger,
ParameterString,)
Processing_instance_count = ParameterInteger(name="ProcessingInstanceCount",default_value=1)

data_input1 = ParameterString(name="data1",default_value=data1,)
data_input2 = ParameterString(name="data2",default_value=data2,)

处理步骤代码 #

process_step = ProcessingStep(     
    name="...",     
    processor=sklearn_processor,     
    inputs=[         
        ProcessingInput(source=data1, destination="/opt/ml/processing/input"),        
        ProcessingInput(source=data2, destination="/opt/ml/processing/input"),],     
    outputs=[         
        ProcessingOutput(output_name="...", source="/opt/ml/processing/output"),],    
    code=preprocess_script_uri,     
)

管道定义代码 #

pipeline_name = f"..."       
pipeline = Pipeline(     
    name=pipeline_name,     
    parameters=[         
        Processing_instance_count,         
        data_input1,         
        data_input2,     ],     
    steps=[process_step],     
)

问题根源与修复方案 #

1. 重复目标路径导致数据覆盖 #

两个ProcessingInput都指向同一个目标路径/opt/ml/processing/input，这会导致后一个输入覆盖前一个的文件，或引发文件系统冲突，最终让预处理脚本找不到完整的数据源，触发AlgorithmError。

修复：给每个输入分配独立子路径：

inputs=[         
    ProcessingInput(source=data_input1, destination="/opt/ml/processing/input/data1"),        
    ProcessingInput(source=data_input2, destination="/opt/ml/processing/input/data2"),
]

同时修改预处理脚本，从对应的子路径读取两个数据源进行合并。

2. 参数引用错误导致Job创建失败 #

处理步骤中直接使用了原始变量data1/data2，而非定义好的参数对象data_input1/data_input2，管道无法正确解析参数化的输入路径，从而引发CreateProcessingJob调用失败。

修复：修正ProcessingInput的source参数，使用定义好的参数对象：

inputs=[         
    ProcessingInput(source=data_input1, destination="/opt/ml/processing/input/data1"),        
    ProcessingInput(source=data_input2, destination="/opt/ml/processing/input/data2"),
]

额外验证点 #

• 确认预处理脚本已适配新的子路径，没有硬编码旧路径
• 检查SageMaker执行角色对两个数据源的存储位置（如S3）均有读取权限
• 验证两个数据源的文件格式一致，避免脚本因格式不兼容崩溃

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