是否可以选择我想要存储管道运行文件的S3目录？

Question

我的管道运行目录正在我的默认存储桶目录中创建，并且它变得混乱。 我一直在尝试在 s3 默认存储桶中设置一个子目录来存储所有管道运行目录（预处理、训练、评估、解释...），但我还没有成功。有人可以帮我做到这一点吗？

我尝试将步骤名称从 CrossPreprocess-Data 更改为 sagemaker/cross-project/CrossPreprocess-Data 并且它有效。它在 sagemaker/cross-project 目录中创建了我的管道目录

    step_process = ProcessingStep(
        name="sagemaker/cross-project/CrossPreprocess-Data",
        processor=sklearn_processor,
        outputs=[
            ProcessingOutput(output_name="train", source="/opt/ml/processing/train"),
            ProcessingOutput(output_name="validation", source="/opt/ml/processing/val"),
            ProcessingOutput(output_name="test", source="/opt/ml/processing/test"),
            ProcessingOutput(output_name="metafiles", source="/opt/ml/processing/metafiles"),
        ],
        code=os.path.join(BASE_DIR, "preprocess.py"),
        job_arguments=["--input-data", input_data, "--run-datetime", run_datetime, "--project-name", project_name],
    )

但是，在下一步中，我收到此错误，并且无法继续执行

ClientError: An error occurred (ValidationException) when calling the UpdatePipeline operation: Unable to parse pipeline definition. Invalid property reference 'Steps.sagemaker/cross-project/CrossPreprocess-Data.ProcessingOutputConfig.Outputs['train'].S3Output.S3Uri' in GetFunction definition.

Answer 1

您可以在

ProcessingOutput

中设置目标位置，以指定文件应保存在 S3 存储桶中的位置。

https://sagemaker.readthedocs.io/en/stable/api/training/processing.html#sagemaker.processing.ProcessingOutput

Answer 2

需要采取几个步骤来管理 sagemaker Pipeline 运行期间上传文件的位置：

1. Outputs：指定

   ProcessingStep

   的输出时，将

   destination

   参数添加到

   ProcessingOutput

   来告诉作业将最终输出上传到 s3 上的何处

ProcessingOutput(
    output_name="train", 
    source="/opt/ml/processing/train", 
    destination=f"s3://{sagemaker_session.default_bucket()}/project_folder/outputs"
)

请注意，如果您使用相同的目标重新运行作业，输出将覆盖该目标之前的输出，除非将作业 ID 等唯一标识符附加到目标路径。

2. 输入：我们还需要管理输入的上传位置。我们可以通过在传递 PipelineSession

    的地方向 

   sagemaker_session 对象

    添加 

   bucket 前缀来实现此目的。通过将

   default_bucket_prefix

   设置为您要使用的默认存储桶下的任何新文件夹来执行此操作：

实例化处理器以稍后传递到

ProcessingStep

...

processor = ScriptProcessor(
    command=['python3'],
    image_uri="[your aws account id].dkr.ecr.[region].amazonaws.com/[ecr repository name]",
    role="[role arn]",
    instance_count=1,
    instance_type='ml.m5.4xlarge',
    sagemaker_session=PipelineSession(default_bucket_prefix="pipeline-job-runs")
)

或者在实例化估计器以稍后传递到

TrainingStep

...

时

estimator = PyTorch(
    entry_point="fine_tune.py",
    source_dir="src/train",
    base_job_name="project-folder/training",
    instance_type="ml.g5.12xlarge",
    instance_count=1,
    role="[role arn]",
    py_version="py310",
    framework_version="2.1.0",
    hyperparameters={
        'model_id': model_id,
        'per_device_train_batch_size': 1,
    },
    sagemaker_session=PipelineSession(default_bucket_prefix="pipeline-job-runs")
)