使用本地模式运行管道

SageMaker 管道本地模式是一种在托管 SageMaker 服务上执行管道之前测试训练、处理和推理脚本以及管道参数的运行时兼容性的简便方法。通过使用本地模式，您可以使用较小的数据集在本地测试 SageMaker 管道。这样可以快速轻松地调试用户脚本和管道定义本身中的错误，而不会产生使用托管服务的成本。以下主题向您展示如何在本地定义和运行管道。

管道本地模式在幕后利用SageMaker 作业本地模式。这是 SageMaker Python 中的一项功能SDK，允许你使用 Docker 容器在本地运行 SageMaker 内置或自定义镜像。管道本地模式建立在 SageMaker 作业本地模式之上。因此，您将会看到与单独运行这些作业时相同的结果。例如，本地模式仍使用 Amazon S3 上传模型构件和处理输出。如果要将本地作业生成的数据存储在本地磁盘上，可以使用本地模式中提到的设置。

管道本地模式目前支持以下步骤类型：

托管管道服务允许使用并行配置并行执行多个步骤，而本地管道执行程序则是按顺序运行步骤。因此，本地管道的总体执行性能可能比在云上运行的管道差，这主要取决于数据集的大小、算法以及本地计算机的性能。另请注意，在本地模式下运行的流水线不会记录在SageMaker 实验中。

为了在本地执行管道，与管道步骤和管道本身关联的 sagemaker_session 字段必须是 LocalPipelineSession 类型。以下示例说明如何定义要在本地执行的 SageMaker 管道。

from sagemaker.workflow.pipeline_context import LocalPipelineSession
from sagemaker.pytorch import PyTorch
from sagemaker.workflow.steps import TrainingStep
from sagemaker.workflow.pipeline import Pipeline

local_pipeline_session = LocalPipelineSession()

pytorch_estimator = PyTorch(
    sagemaker_session=local_pipeline_session,
    role=sagemaker.get_execution_role(),
    instance_type="ml.c5.xlarge",
    instance_count=1,
    framework_version="1.8.0",
    py_version="py36",
    entry_point="./entry_point.py",
)

step = TrainingStep(
    name="MyTrainingStep",
    step_args=pytorch_estimator.fit(
        inputs=TrainingInput(s3_data="s3://amzn-s3-demo-bucket/my-data/train"),
    )
)

pipeline = Pipeline(
    name="MyPipeline",
    steps=[step],
    sagemaker_session=local_pipeline_session
)

pipeline.create(
    role_arn=sagemaker.get_execution_role(), 
    description="local pipeline example"
)

// pipeline will execute locally
execution = pipeline.start()

steps = execution.list_steps()

training_job_name = steps['PipelineExecutionSteps'][0]['Metadata']['TrainingJob']['Arn']

step_outputs = pipeline_session.sagemaker_client.describe_training_job(TrainingJobName = training_job_name)

准备好在托管 Pipelin SageMaker es 服务上执行管道后，可以通过将前面的代码片段LocalPipelineSession中的替换为PipelineSession（如以下代码示例所示），然后重新运行代码来实现。

from sagemaker.workflow.pipeline_context import PipelineSession

pipeline_session = PipelineSession()