如何通过SageMaker在多EC2实例上调度带自定义参数的脚本

你需要的是创建多个独立的SageMaker Processing Job，每个Job使用1台EC2实例，并传入不同的自定义参数——注意instance_count=1是单Job用1台实例，而非让多台实例跑同一个任务。下面是具体实现方案：

核心思路 #

把需要运行的自定义参数整理成列表，循环遍历每个参数，为每个参数创建独立的Processing Job：

• 每个Job可指定不同的EC2实例类型
• 每个Job传入专属的自定义参数
• 为每个Job设置唯一名称，避免冲突
• 隔离不同Job的S3输出路径，防止结果覆盖

修改后的完整代码 #

import boto3, os, sagemaker
from sagemaker.processing import ScriptProcessor, ProcessingInput, ProcessingOutput
from sagemaker import get_execution_role

sagemaker_session = sagemaker.Session()
default_s3_bucket = sagemaker_session.default_bucket()
print(default_s3_bucket)

# 上传共用的输入数据（若不同Job需不同输入，可在循环内单独上传）
input_data = sagemaker_session.upload_data(path='./data_files', 
                                           bucket=default_s3_bucket)

role = get_execution_role()

# 定义任务参数列表，每个元素对应一组参数和实例配置
task_parameters = [
    {"example_arg": "value1", "instance_type": "ml.c5.xlarge"},
    {"example_arg": "value2", "instance_type": "ml.c5.2xlarge"},
    {"example_arg": "value3", "instance_type": "ml.m5.large"}
]

# 循环创建并运行每个独立任务
for idx, params in enumerate(task_parameters):
    # 为当前任务创建ScriptProcessor实例
    script_processor = ScriptProcessor(
        command=['python3'],
        image_uri='ecr/docker_image:latest',
        role=role,
        instance_count=1,  # 每个任务用1台独立实例
        base_job_name=f'run-processing-task-{idx}',  # 唯一任务名称
        instance_type=params["instance_type"]
    )
    
    # 为当前任务设置独立输出路径
    output_location = os.path.join('s3://', default_s3_bucket, f'processing_output/task_{idx}')
    
    # 启动任务并传入自定义参数
    script_processor.run(
        code='process_script.py',
        arguments=['--example_argument', params["example_arg"]],
        inputs=[ProcessingInput(source=input_data, destination='/opt/ml/processing/input')],
        outputs=[ProcessingOutput(source='/opt/ml/processing/output', destination=output_location)]
    )

关键说明 #

• 参数与实例绑定：task_parameters列表可灵活为每个任务指定不同参数和实例类型，适配多样化需求
• 任务唯一性：通过base_job_name添加索引后缀，避免SageMaker任务重名报错
• 输出隔离：每个任务的S3输出路径添加专属标识，防止不同任务的输出文件互相覆盖
• 执行方式控制：默认run()为同步执行，完成一个任务后再启动下一个；若需并行执行，可添加wait=False参数，之后通过SageMaker控制台或API监控任务状态

扩展优化 #

• 若不同任务需不同输入数据，可在循环内部单独调用upload_data()上传对应资源
• 可添加异常捕获逻辑，处理任务运行失败的情况并实现重试机制
• 针对大量任务，可使用SageMaker Pipeline编排，实现更复杂的依赖管理和调度策略

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