パイプラインを実行する

JSON パイプライン定義を調べて、適切に形成されていることを確認します。

import json

json.loads(pipeline.definition())

パイプライン定義を Pipelines サービスに送信してパイプラインが存在しない場合はパイプラインを作成するか、パイプラインが存在しない場合はパイプラインを更新します。渡されたロールは、パイプラインによってステップで定義されたすべてのジョブを作成するために使用されます。

pipeline.upsert(role_arn=role)

パイプラインの実行を開始します。

execution = pipeline.start()

パイプラインの実行ステータスを書き留め、パイプラインが正常に作成され、開始されたことを確認します。

execution.describe()

実行が終了するまで待ちます。

execution.wait()

実行ステップとそのステータスをリストします。

execution.list_steps()

出力は次のようになります。

[{'StepName': 'AbaloneTransform',
  'StartTime': datetime.datetime(2020, 11, 21, 2, 41, 27, 870000, tzinfo=tzlocal()),
  'EndTime': datetime.datetime(2020, 11, 21, 2, 45, 50, 492000, tzinfo=tzlocal()),
  'StepStatus': 'Succeeded',
  'CacheHitResult': {'SourcePipelineExecutionArn': ''},
  'Metadata': {'TransformJob': {'Arn': 'arn:aws:sagemaker:us-east-2:111122223333:transform-job/pipelines-cfvy1tjuxdq8-abalonetransform-ptyjoef3jy'}}},
 {'StepName': 'AbaloneRegisterModel',
  'StartTime': datetime.datetime(2020, 11, 21, 2, 41, 26, 929000, tzinfo=tzlocal()),
  'EndTime': datetime.datetime(2020, 11, 21, 2, 41, 28, 15000, tzinfo=tzlocal()),
  'StepStatus': 'Succeeded',
  'CacheHitResult': {'SourcePipelineExecutionArn': ''},
  'Metadata': {'RegisterModel': {'Arn': 'arn:aws:sagemaker:us-east-2:111122223333:model-package/abalonemodelpackagegroupname/1'}}},
 {'StepName': 'AbaloneCreateModel',
  'StartTime': datetime.datetime(2020, 11, 21, 2, 41, 26, 895000, tzinfo=tzlocal()),
  'EndTime': datetime.datetime(2020, 11, 21, 2, 41, 27, 708000, tzinfo=tzlocal()),
  'StepStatus': 'Succeeded',
  'CacheHitResult': {'SourcePipelineExecutionArn': ''},
  'Metadata': {'Model': {'Arn': 'arn:aws:sagemaker:us-east-2:111122223333:model/pipelines-cfvy1tjuxdq8-abalonecreatemodel-jl94rai0ra'}}},
 {'StepName': 'AbaloneMSECond',
  'StartTime': datetime.datetime(2020, 11, 21, 2, 41, 25, 558000, tzinfo=tzlocal()),
  'EndTime': datetime.datetime(2020, 11, 21, 2, 41, 26, 329000, tzinfo=tzlocal()),
  'StepStatus': 'Succeeded',
  'CacheHitResult': {'SourcePipelineExecutionArn': ''},
  'Metadata': {'Condition': {'Outcome': 'True'}}},
 {'StepName': 'AbaloneEval',
  'StartTime': datetime.datetime(2020, 11, 21, 2, 37, 34, 767000, tzinfo=tzlocal()),
  'EndTime': datetime.datetime(2020, 11, 21, 2, 41, 18, 80000, tzinfo=tzlocal()),
  'StepStatus': 'Succeeded',
  'CacheHitResult': {'SourcePipelineExecutionArn': ''},
  'Metadata': {'ProcessingJob': {'Arn': 'arn:aws:sagemaker:us-east-2:111122223333:processing-job/pipelines-cfvy1tjuxdq8-abaloneeval-zfraozhmny'}}},
 {'StepName': 'AbaloneTrain',
  'StartTime': datetime.datetime(2020, 11, 21, 2, 34, 55, 867000, tzinfo=tzlocal()),
  'EndTime': datetime.datetime(2020, 11, 21, 2, 37, 34, 34000, tzinfo=tzlocal()),
  'StepStatus': 'Succeeded',
  'CacheHitResult': {'SourcePipelineExecutionArn': ''},
  'Metadata': {'TrainingJob': {'Arn': 'arn:aws:sagemaker:us-east-2:111122223333:training-job/pipelines-cfvy1tjuxdq8-abalonetrain-tavd6f3wdf'}}},
 {'StepName': 'AbaloneProcess',
  'StartTime': datetime.datetime(2020, 11, 21, 2, 30, 27, 160000, tzinfo=tzlocal()),
  'EndTime': datetime.datetime(2020, 11, 21, 2, 34, 48, 390000, tzinfo=tzlocal()),
  'StepStatus': 'Succeeded',
  'CacheHitResult': {'SourcePipelineExecutionArn': ''},
  'Metadata': {'ProcessingJob': {'Arn': 'arn:aws:sagemaker:us-east-2:111122223333:processing-job/pipelines-cfvy1tjuxdq8-abaloneprocess-mgqyfdujcj'}}}]

パイプラインの実行が完了したら、Amazon S3 から結果の evaluation.json ファイルをダウンロードして、レポートを確認します。

evaluation_json = sagemaker.s3.S3Downloader.read_file("{}/evaluation.json".format(
    step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"]
))
json.loads(evaluation_json)

パイプラインの実行を作成します。これにより、モデル承認ステータスのオーバーライドが「承認済み」に設定された状態で別のパイプライン実行が開始されます。つまり、RegisterModelステップによって生成されたモデルパッケージバージョンは、 SageMaker Projects などの CI/CD パイプラインを介して自動的にデプロイできる状態になります。詳細については、「MLOps SageMaker プロジェクトによる自動化」を参照してください。

execution = pipeline.start(
    parameters=dict(
        ModelApprovalStatus="Approved",
    )
)

実行が終了するまで待ちます。

execution.wait()

実行ステップとそのステータスをリストします。

execution.list_steps()

パイプラインの実行が完了したら、Amazon S3 から結果の evaluation.json ファイルをダウンロードして、レポートを確認します。

evaluation_json = sagemaker.s3.S3Downloader.read_file("{}/evaluation.json".format(
    step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"]
))
json.loads(evaluation_json)

パイプラインの実行を停止します。

execution.stop()

パイプラインを削除します。

pipeline.delete()