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

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

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

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

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

パイプラインをローカルで実行するには、パイプラインステップとパイプライン自体に関連付けられた sagemaker_session フィールドが LocalPipelineSession 型と一致している必要があります。次の例は、ローカルで実行される SageMaker AI パイプラインを定義する方法を示しています。

from sagemaker.core.workflow.pipeline_context import LocalPipelineSession
from sagemaker.train import ModelTrainer
from sagemaker.train.configs import InputData, SourceCode
from sagemaker.train.configs import Compute
from sagemaker.mlops.workflow.steps import TrainingStep
from sagemaker.mlops.workflow.pipeline import Pipeline
from sagemaker.core.helper.session_helper import get_execution_role
from sagemaker.core import image_uris

local_pipeline_session = LocalPipelineSession()

training_image = image_uris.retrieve(
    framework="pytorch", region=region, version="1.8.0",
    py_version="py36", instance_type="ml.c5.xlarge",
    image_scope="training"
)

pytorch_model_trainer = ModelTrainer(
    role=get_execution_role(),
    compute=Compute(
        instance_type="ml.c5.xlarge",
        instance_count=1,
    ),
    training_image=training_image,
    source_code=SourceCode(entry_script="./entry_point.py"),
)

step = TrainingStep(
    name="MyTrainingStep",
    step_args=pytorch_model_trainer.train(
        input_data_config=[InputData(channel_name="training", data_source="s3://amzn-s3-demo-bucket/my-data/train")],
    )
)

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

pipeline.create(
    role_arn=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 = local_pipeline_session.sagemaker_client.describe_training_job(TrainingJobName = training_job_name)

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

from sagemaker.workflow.pipeline_context import PipelineSession

pipeline_session = PipelineSession()