ローカルモードを使用してパイプラインを実行する

SageMaker Pipelines ローカルモードは、マネージド SageMaker サービスでパイプラインを実行する前に、トレーニング、処理、推論スクリプト、パイプラインパラメータのランタイム互換性を簡単にテストできます。ローカルモードを使用すると、小さいデータセットを使用して SageMaker パイプラインをローカルでテストできます。これにより、マネージドサービスの使用コストをかけずに、ユーザースクリプトやパイプライン定義自体のエラーをすばやく簡単にデバッグできます。次のトピックでは、パイプラインをローカルで定義して実行する方法について説明します。

Pipelines ローカルモードは、フードの下にSageMaker ジョブのローカルモードを使用します。これは、Docker コンテナを使用して SageMaker 組み込みイメージまたはカスタムイメージをローカルでSDK実行できる SageMaker Python の機能です。Pipelines ローカルモードは、 SageMaker ジョブのローカルモードの上に構築されています。そのため、これらのジョブを別々に実行した場合と同じ結果が得られることが期待できます。例えば、ローカルモードでは引き続き Amazon S3 を使用してモデルアーティファクトをアップロードし、出力を処理します。ローカルジョブによって生成されたデータをローカルディスクに保存したい場合は、「Local Mode」で説明されている設定を使用できます。

パイプラインローカルモードは現在、以下のステップタイプをサポートしています。

並列処理設定を使用して複数のステップを並行して実行できるマネージドパイプラインサービスとは対照的に、ローカルのパイプラインエグゼキューターはステップを順番に実行します。そのため、ローカルパイプラインの全体的な実行パフォーマンスはクラウド上で実行されるパイプラインよりも劣る可能性があります。このパフォーマンスは主に、データセットのサイズ、アルゴリズム、ローカルコンピューターの処理能力に依存します。また、ローカルモードでのパイプラインの実行は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)

マネージド SageMaker パイプラインサービスでパイプラインを実行する準備ができたら、前のコードスニペットLocalPipelineSessionの を PipelineSession (次のコードサンプルに示すように) に置き換え、コードを再実行することで実行できます。

from sagemaker.workflow.pipeline_context import PipelineSession

pipeline_session = PipelineSession()