运行管道 - Amazon SageMaker
AWS 文档中描述的 AWS 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 AWS 服务入门

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

运行管道

使用 SageMaker Python 开发工具包创建管道定义后,您可以将其提交到 SageMaker 以开始执行。以下教程介绍如何提交管道,启动执行,检查执行结果,以及删除管道。

Prerequisites

本教程要求以下项目:

  • 笔记本实例。SageMaker 

  • 管道定义。SageMaker Pipelines本教程假定您使用的是通过完成 定义管道 教程创建的管道定义。

步骤 1:启动管道

首先,您需要启动管道。

启动管道

  1. 检查 JSON 管道定义以确保其格式正确。

    import json json.loads(pipeline.definition())
  2. 将管道定义提交到 SageMaker Pipelines 服务,以便创建管道(如果管道不存在)或更新管道(如果管道不存在)。传入的角色由 SageMaker Pipelines 用来创建步骤中定义的所有任务。

    pipeline.upsert(role_arn=role)
  3. 启动管道执行。

    execution = pipeline.start()

步骤 2:检查管道执行

接下来,您需要检查管道执行。

检查管道执行

  1. 描述管道执行状态以确保其已成功创建和启动。

    execution.describe()
  2. 等待执行完成。

    execution.wait()
  3. 列出执行步骤及其状态。

    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'}}}]
  4. 管道执行完成后,从 evaluation.json 下载生成的 Amazon S3 文件以检查报告。

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

步骤 3:覆盖管道执行的默认参数

您可以通过指定不同的管道参数覆盖默认值来运行管道的其他执行。

覆盖默认参数

  1. 创建管道执行。这将在计算优化型实例类型上启动另一个管道执行,并将模型批准状态设置为“Approved (已批准)”。这意味着,RegisterModel 步骤生成的模型包版本会自动准备好通过 CI/CD 管道进行部署,例如通过 SageMaker 项目进行部署。有关更多信息,请参阅通过SageMaker项目自动完成 MLOps

    execution = pipeline.start( parameters=dict( ProcessingInstanceType="ml.c5.xlarge", ModelApprovalStatus="Approved", ) )
  2. 等待执行完成。

    execution.wait()
  3. 列出执行步骤及其状态。

    execution.list_steps()
  4. 管道执行完成后,从 evaluation.json 下载生成的 Amazon S3 文件以检查报告。

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

步骤 4:停止和删除管道执行

完成您的管道后,您可以停止任何持续执行并删除管道。

停止和删除管道执行

  1. 停止管道执行。

    execution.stop()
  2. 删除管道。

    pipeline.delete()