本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # SageMaker AI 工作流程 工作流 在扩展机器学习 (ML) 操作时，您可以使用 Amazon A SageMaker I 完全托管的工作流程服务来实施机器学习生命周期的持续集成和部署 (CI/CD) 实践。利用 Pipelines SDK，您可以选择管道步骤并将其集成到统一的解决方案中，从而实现从数据准备到模型部署的模型构建流程自动化。对于基于 Kubernetes 的架构，你可以在 Kubernetes 集群上安装 SageMaker 人工智能运算符，使用 Kubernetes API 和命令行 Kubernetes 工具（例如）在本地创建 SageMaker 人工智能作业。`kubectl`借助 Kubeflow 管道的 SageMaker AI 组件，您可以通过 Kubeflow 管道创建和监控原生 SageMaker AI 作业。可从 Kubeflow Pipelines 用户界面访问 SageMaker 来自 AI 的作业参数、状态和输出。最后，如果要安排批处理作业，则可以使用 Amazon Batch 作业队列集成或基于 Jupyter 笔记本的工作流服务，按照您定义的计划启动独立运行或常规运行。 总而言之， SageMaker AI 提供了以下工作流程技术： + [管道](pipelines.md)：用于构建和管理机器学习管道的工具。 + [Kubernetes 编排](kubernetes-workflows.md): Kubernetes 集群的 SageMaker AI 自定义运算符和 Kubeflow Pipelines 的组件。 + [SageMaker 笔记本职位](notebook-auto-run.md)：按需或按计划非交互式批量运行 Jupyter 笔记本。 您还可以利用与 SageMaker AI 集成的其他服务来构建您的工作流程。这些选项包含以下服务： + [气流工作流程](https://sagemaker.readthedocs.io/en/stable/workflows/airflow/index.html)：导出 SageMaker APIs 用于创建和管理 Airflow 工作流程的配置。 + [Amazon Step Functions](https://sagemaker.readthedocs.io/en/stable/workflows/step_functions/index.html)：Python 中的多步机器学习工作流程，无需单独配置资源即可编排 SageMaker AI 基础架构。 + [Amazon Batch](https://docs.amazonaws.cn/batch/latest/userguide/getting-started-sagemaker.html)：将 SageMaker AI 训练作业提交到 Amazon Batch 作业队列，在那里你可以对作业进行优先级排序和安排，使其在计算环境中运行。 有关管理 SageMaker 训练和推理的更多信息，请参阅 [Amaz SageMaker on Python 软件开发工具包工作流程](https://sagemaker.readthedocs.io/en/stable/workflows/index.html)。 **Topics** + [ # 管道 ](pipelines.md) + [ # Kubernetes 编排 ](kubernetes-workflows.md) + [ # SageMaker 笔记本职位 ](notebook-auto-run.md) + [ # 安排您的 ML 工作流程 ](workflow-scheduling.md) + [ # Amazon Batch 支持 A SageMaker I 训练作业 ](training-job-queues.md) # 管道 ML Pipelines Amaz SageMaker on Pipelines 是一项专门构建的工作流程编排服务，用于自动进行机器学习 (ML) 开发。 与其他 Amazon 工作流程产品相比，管道具有以下优势： **自动扩展无服务器基础架构**您无需管理底层编排基础架构即可运行 Pipelines，这使您可以专注于核心机器学习任务。 SageMaker AI 会根据您的机器学习工作负载的要求自动配置、扩展和关闭管道编排计算资源。 **直观的用户体验**可以通过您选择的界面创建和管理管道：可视化编辑器 APIs、SDK 或 JSON。您可以 drag-and-drop在 Amazon SageMaker Studio 可视化界面中执行各种机器学习步骤来创作您的管道。下面的截图显示了 Studio 管道可视化编辑器。 ![\[Studio 中流水线的可视 drag-and-drop界面屏幕截图。\]](http://docs.amazonaws.cn/sagemaker/latest/dg/images/pipelines/pipelines-studio-overview.png) 如果您更喜欢以编程方式管理机器学习工作流程， SageMaker Python SDK 可提供高级编排功能。有关更多信息，请参阅 Python 软件开发工具包[ SageMaker 文档中的 Amaz](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html) SageMaker on Pipelines。 **Amazon 集成** Pipelines 提供与所有 SageMaker AI 功能和其他 Amazon 服务的无缝集成，可自动执行数据处理、模型训练、微调、评估、部署和监控作业。您可以将 SageMaker AI 功能整合到您的管道中，并使用深度链接在这些功能中进行导航，从而大规模创建、监控和调试您的机器学习工作流程。 **降低成本**使用 Pipelines，您只需为 SageMaker Studio 环境和由管道编排的底层作业（例如 SageMaker 训练、 SageMaker 处理、 SageMaker AI 推理和 Amazon S3 数据存储）付费。 **可审计性和任务流水线追踪功能**：有了 Pipelines，您可以使用内置的版本控制功能来跟踪管道更新与执行的历史记录。Amazon SageMaker ML Lineage Tracking 可帮助您分析 end-to-end机器学习开发生命周期中的数据源和数据使用者。 **Topics** + [ # 管道概述 ](pipelines-overview.md) + [ # Pipelines 操作 ](pipelines-build.md) # 管道概述 管道概述 Amazon SageMaker AI 管道是定向无环图 (DAG) 中一系列相互关联的步骤，这些步骤是使用 drag-and-drop UI 或 Pipelin [es SDK](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html) 定义的。您还可以使用[管道定义 JSON 模式](https://aws-sagemaker-mlops.github.io/sagemaker-model-building-pipeline-definition-JSON-schema/)构建管道。此 DAG JSON 定义提供了管道中每个步骤的要求和关系信息。管道 DAG 的结构由步骤之间的数据依赖关系决定。当一个步骤的输出属性作为输入传递给另一个步骤时，就会产生这些数据依赖关系。下图是管道 DAG 的示例： ![\[\]](http://docs.amazonaws.cn/sagemaker/latest/dg/images/pipeline-full.png) **示例 DAG 包括以下步骤：** 1. `AbaloneProcess` 是[处理](https://docs.amazonaws.cn/sagemaker/latest/dg/build-and-manage-steps.html#step-type-processing)步骤的一个实例，在用于训练的数据上运行预处理脚本。例如，脚本可以填充缺失值，对数值数据进行规范化处理，或将数据分成训练数据集、验证数据集和测试数据集。 1. `AbaloneTrain` 是[训练](https://docs.amazonaws.cn/sagemaker/latest/dg/build-and-manage-steps.html#step-type-training)步骤的一个实例，用于配置超参数，并根据预处理的输入数据训练模型。 1. `AbaloneEval` 是[处理](https://docs.amazonaws.cn/sagemaker/latest/dg/build-and-manage-steps.html#step-type-processing)步骤的另一个实例，用于评估模型的准确性。本步骤展示了一个数据依赖的例子 - 本步骤使用 `AbaloneProcess` 的测试数据集输出。 1. `AbaloneMSECond`是 Condit [i](https://docs.amazonaws.cn/sagemaker/latest/dg/build-and-manage-steps.html#step-type-condition) on 步骤的一个实例，在本例中，该步骤进行检查以确保模型评估的 mean-square-error结果低于特定限制。如果模型不符合标准，管道运行就会停止。 1. 管道运行按以下步骤进行： 1. `AbaloneRegisterModel`，其中 SageMaker AI 调用[RegisterModel](https://docs.amazonaws.cn/sagemaker/latest/dg/build-and-manage-steps.html#step-type-register-model)步骤将模型作为版本控制的模型包组注册到 Amazon SageMaker 模型注册表中。 1. `AbaloneCreateModel`，其中 SageMaker AI 调用创建模型的[CreateModel](https://docs.amazonaws.cn/sagemaker/latest/dg/build-and-manage-steps.html#step-type-create-model)步骤，为批量转换做准备。在中`AbaloneTransform`， SageMaker AI 调用 “[转换](https://docs.amazonaws.cn/sagemaker/latest/dg/build-and-manage-steps.html#step-type-transform)” 步骤，对您指定的数据集生成模型预测。 以下主题介绍了 Pipelines 的基本概念。有关描述这些概念的实施的教程，请参阅[Pipelines 操作](pipelines-build.md)。 **Topics** + [ # 管道结构和执行 ](build-and-manage-pipeline.md) + [ # IAM 访问管理 ](build-and-manage-access.md) + [ # 为 Pipelines 设置跨账户支持 ](build-and-manage-xaccount.md) + [ # Pipeline 参数 ](build-and-manage-parameters.md) + [ # Pipelines 步骤 ](build-and-manage-steps.md) + [ # Lift-and-shift 使用 @step 装饰器的 Python 代码 ](pipelines-step-decorator.md) + [ # 在步骤之间传递数据 ](build-and-manage-propertyfile.md) + [ # 缓存管道步骤 ](pipelines-caching.md) + [ # 管道步骤的重试策略 ](pipelines-retry-policy.md) + [ # 选择性执行管道步骤 ](pipelines-selective-ex.md) + [ # 基准计算、偏差检测和生命周期以及 Amazon Pi ClarifyCheck pelin SageMaker es 中的 QualityCheck 步骤 ](pipelines-quality-clarify-baseline-lifecycle.md) + [ # 安排管道运行 ](pipeline-eventbridge.md) + [ # 亚马逊 SageMaker 实验集成 ](pipelines-experiments.md) + [ # 使用本地模式运行管道 ](pipelines-local-mode.md) + [ # 对亚马逊 SageMaker 管道进行故障排除 ](pipelines-troubleshooting.md) # 管道结构和执行 结构和执行 **Topics** + [ ## 管线结构 ](#build-and-manage-pipeline-structure) + [ ## 使用并行配置执行管道 ](#build-and-manage-pipeline-execution) ## 管线结构 结构 Amaz SageMaker on Pipelines 实例由`name`、`parameters`、和组成`steps`。`(account, region)` 对中的管道名称必须是唯一的。步骤定义中使用的所有参数都必须在管道中定义。列出的管道步骤会根据彼此之间的数据依赖关系自动确定其执行顺序。管道服务会解析数据依赖 DAG 中各步骤之间的关系，创建一系列执行完成的步骤。以下是管道结构的示例。 **警告** 通过可视化编辑器或 SageMaker AI Python SDK 构建管道时，请勿在管道参数或任何步骤定义字段（例如环境变量）中包含敏感信息。这些字段在未来通过 `DescribePipeline` 请求返回时将可见。 ``` from sagemaker.workflow.pipeline import Pipeline pipeline_name = f"AbalonePipeline" pipeline = Pipeline( name=pipeline_name, parameters=[ processing_instance_type, processing_instance_count, training_instance_type, model_approval_status, input_data, batch_data, ], steps=[step_process, step_train, step_eval, step_cond], ) ``` ## 使用并行配置执行管道 Execution 默认情况下，管线会执行所有可并行运行的步骤。创建或更新管线以及启动或重试管线执行时，您可以使用 `ParallelismConfiguration` 属性来控制此行为。 每个执行都会应用并行配置。例如，如果启动了两个执行，则每个执行最多可以同时运行 50 个步骤，总共可以同时运行 100 个步骤。此外，启动、重试或更新执行时指定的 `ParallelismConfiguration` 优先于管道中定义的并行配置。 **Example 使用 `ParallelismConfiguration` 创建管道执行** ``` pipeline = Pipeline( name="myPipeline", steps=[step_process, step_train] ) pipeline.create(role, parallelism_config={"MaxParallelExecutionSteps": 50}) ``` # IAM 访问管理 访问控制 以下各节描述了亚马逊 SageMaker 管道的 Amazon Identity and Access Management (IAM) 要求。有关如何实施这些权限的示例，请参阅[先决条件](define-pipeline.md#define-pipeline-prereq)。 **Topics** + [ ## 管道角色权限 ](#build-and-manage-role-permissions) + [ ## 管道步骤权限 ](#build-and-manage-step-permissions) + [ ## 使用 Amazon S3 存储桶进行 CORS 配置 ](#build-and-manage-cors-s3) + [ ## 自定义 Pipelines 作业的访问管理 ](#build-and-manage-step-permissions-prefix) + [ ## 自定义对管道版本的访问权限 ](#build-and-manage-step-permissions-version) + [ ## 使用管道的服务控制策略 ](#build-and-manage-scp) ## 管道角色权限 您的管道需要一个 IAM 管道执行角色，该角色将在您创建管道时传递给管道。 您用于创建管道的 SageMaker AI 实例的角色必须具有指定管道执行角色的`iam:PassRole`权限的策略。这是因为实例需要权限才能将您的管道执行角色传递给 Pipelines 服务，以用于创建和运行管道。有关 IAM 角色的更多信息，请参阅 [IAM 角色](https://docs.amazonaws.cn/IAM/latest/UserGuide/id_roles.html)。 您的管道执行角色需要以下权限： + 您可以为管道中的任何 SageMaker AI 任务步骤使用唯一角色或自定义角色（而不是默认使用的管道执行角色）。请确保您的管道执行角色已添加一个策略，该策略具有指定每个角色的`iam:PassRole`权限。 + 管道中每种作业类型的 `Create` 和 `Describe` 权限。 + Amazon S3 使用 `JsonGet` 函数的权限。您可以使用基于资源的策略和基于身份的策略来控制对 Amazon S3 资源的访问。基于资源的策略会应用到 Amazon S3 存储桶，并授予 Pipelines 对存储桶的访问权限。基于身份的策略使您的管道能够从您的账户调用 Amazon S3。有关基于资源的策略和基于身份的策略的更多信息，请参阅[基于身份的策略和基于资源的策略](https://docs.amazonaws.cn/IAM/latest/UserGuide/access_policies_identity-vs-resource.html)。 ``` { "Action": [ "s3:GetObject" ], "Resource": "arn:aws:s3:::/*", "Effect": "Allow" } ``` ## 管道步骤权限 管道包括运行 A SageMaker I 作业的步骤。为了让管道步骤运行这些作业，需要在您的账户中设置一个 IAM 角色，以提供对所需资源的访问权限。此角色由您的管道传递给 SageMaker AI 服务主体。有关 IAM 角色的更多信息，请参阅 [IAM 角色](https://docs.amazonaws.cn/IAM/latest/UserGuide/id_roles.html)。 默认情况下，每个步骤都扮演管道执行角色。您可以选择将不同的角色传递给管道中的任何步骤。这样可以确保每个步骤中的代码不会影响其他步骤中使用的资源，除非管道定义中指定的两个步骤之间存在直接关系。您可以在为步骤定义处理器或估算器时传递这些角色。有关如何在这些定义中包含这些角色的示例，请参阅 [SageMaker AI Python SDK 文档](https://sagemaker.readthedocs.io/en/stable/overview.html#using-estimators)。 ## 使用 Amazon S3 存储桶进行 CORS 配置 为确保映像以可预测的方式从 Amazon S3 存储桶导入到管道中，必须向从中导入映像的 Amazon S3 存储桶添加 CORS 配置。此部分说明了如何为 Amazon S3 存储桶设置所需的 CORS 配置。Pipelines 所需的 XML `CORSConfiguration` 与[输入映像数据的 CORS 要求](sms-cors-update.md)中的不同，不过您仍可借助其中的信息，进一步了解 Amazon S3 存储桶的 CORS 要求。 对托管映像的 Amazon S3 存储桶使用以下 CORS 配置代码。有关如何配置 CORS 的说明，请参阅《Amazon Simple Storage Service 用户指南》中的[配置跨源资源共享（CORS）](https://docs.amazonaws.cn/AmazonS3/latest/user-guide/add-cors-configuration.html)。如果您使用 Amazon S3 控制台将策略添加到存储桶，则必须使用 JSON 格式。 **JSON** ``` [ { "AllowedHeaders": [ "*" ], "AllowedMethods": [ "PUT" ], "AllowedOrigins": [ "*" ], "ExposeHeaders": [ "Access-Control-Allow-Origin" ] } ] ``` **XML** ``` * * PUT Access-Control-Allow-Origin ``` 下面的 GIF 演示了 Amazon S3 文档中使用 Amazon S3 控制台添加 CORS 头策略的说明。 ![\[关于如何使用 Amazon S3 管理控制台添加 CORS 标头策略的图片。\]](http://docs.amazonaws.cn/sagemaker/latest/dg/images/sms/gifs/cors-config.gif) ## 自定义 Pipelines 作业的访问管理 您可以进一步自定义 IAM 策略，以便组织中的选定成员可以运行任意或所有管道步骤。例如，您可以向某些用户授予创建训练作业的权限，向另一组用户授予创建处理作业的权限，以及向所有用户授予运行其余步骤的权限。要使用此特征，请选择一个以您的作业名称为前缀的自定义字符串。您的管理员在允许 ARNs 的前面加上前缀，而您的数据科学家则在管道实例化中包含此前缀。由于获允许用户的 IAM 策略包含带有指定前缀的作业 ARN，因此您的管道步骤的后续作业具有继续操作所需的权限。作业前缀默认情况下关闭，您必须在 `Pipeline` 类中启用此选项才能使用它。 对于关闭了前缀的作业，作业名称的格式如图所示，该名称由下表中所述的字段串联而成： `pipelines----` | 字段 | 定义 | | --- | --- | | pipelines | 静态字符串，始终前置。此字符串将管道编排服务标识为作业的来源。 | | executionId | 正在运行的管道实例的随机缓冲区。 | | stepNamePrefix | 用户指定的步骤名称（在管道步骤的 `name` 参数中给出），限于前 20 个字符。 | | entityToken | 随机令牌，用于确保步骤实体的幂等性。 | | failureCount | 当前为完成作业而尝试重试的次数。 | 在这种情况下，作业名称前面没有自定义前缀，相应的 IAM 策略必须与此字符串匹配。 对于开启作业前缀的用户，底层作业名称采用以下形式，自定义前缀指定为 `MyBaseJobName`： **-**-**-** 自定义前缀取代了静态`pipelines`字符串，以帮助您缩小可以作为管道一部分运行 SageMaker AI 作业的用户的选择范围。 **前缀长度限制** 作业名称具有特定于各个管道步骤的内部长度限制。此约束还限定了允许的前缀长度。前缀长度要求如下： | 管道步骤 | 前缀长度 | | --- | --- | | `[TrainingStep](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#trainingstep)`, `[ModelStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#step-collections)`, `[TransformStep](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#transformstep)`, `[ProcessingStep](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#processingstep)`, `[ClarifyCheckStep](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#clarifycheckstep)`, `[QualityCheckStep](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#qualitycheckstep)`, `[RegisterModelStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#step-collections)` | 38 | | `[TuningStep](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#tuningstep)`, `[AutoML](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#automlstep)` | 6 | ### 将作业前缀应用于 IAM 策略 您的管理员创建 IAM 策略，允许具有特定前缀的用户创建作业。以下示例策略允许数据科学家在使用 `MyBaseJobName` 前缀的情况下创建训练作业。 ``` { "Action": "sagemaker:CreateTrainingJob", "Effect": "Allow", "Resource": [ "arn:aws:sagemaker:region:account-id:*/MyBaseJobName-*" ] } ``` ### 将作业前缀应用于管道实例化 您可以使用作业实例类的 `*base_job_name` 参数来指定前缀。 **注意** 创建管道步骤之前，您可以将带有 `*base_job_name` 参数的作业前缀传递给作业实例。此作业实例包含使作业作为管道中的一个步骤运行的必要信息。该参数因所使用的作业实例而异。以下列表显示了每种管道步骤类型应使用的参数： 对于 `[Estimator](https://sagemaker.readthedocs.io/en/stable/api/training/estimators.html)` (`[TrainingStep](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#trainingstep)`)、`[Processor](https://sagemaker.readthedocs.io/en/stable/api/training/processing.html)` (`[ProcessingStep](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#processingstep)`) 和 `[AutoML](https://sagemaker.readthedocs.io/en/stable/api/training/automl.html)` (`[AutoMLStep](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#automlstep)`) 类，使用 `base_job_name` 对于 `[Tuner](https://sagemaker.readthedocs.io/en/stable/api/training/tuner.html)` 类 (`[TuningStep](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#tuningstep)`)，使用 `tuning_base_job_name` 对于 `[Transformer](https://sagemaker.readthedocs.io/en/stable/api/inference/transformer.html)` 类 (`[TransformStep](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#transformstep)`)，使用 `transform_base_job_name` 对于 `[QualityCheckStep](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#qualitycheckstep)` (Quality Check) 和 `[ClarifyCheckstep](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#clarifycheckstep)` (Clarify Check) 类，使用 `[CheckJobConfig](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#checkjobconfig)` 的 `base_job_name` 对于 `[Model](https://sagemaker.readthedocs.io/en/stable/api/inference/model.html)` 类，使用的参数取决于您在将结果传递给 `[ModelStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#step-collections)` 之前是运行 `create` 还是 `register` 如果您调用 `create`，则自定义前缀来自您构造模型（即 `Model(name=)`）时的 `name` 参数 如果您调用 `register`，则自定义前缀来自您调用 `register`（即 `my_model.register(model_package_name=)`）时的 `model_package_name` 参数 以下示例说明了如何为新的训练作业实例指定前缀。 ``` # Create a job instance xgb_train = Estimator( image_uri=image_uri, instance_type="ml.m5.xlarge", instance_count=1, output_path=model_path, role=role, subnets=["subnet-0ab12c34567de89f0"], base_job_name="MyBaseJobName" security_group_ids=["sg-1a2bbcc3bd4444e55"], tags = [ ... ] encrypt_inter_container_traffic=True, ) # Attach your job instance to a pipeline step step_train = TrainingStep( name="TestTrainingJob", estimator=xgb_train, inputs={ "train": TrainingInput(...), "validation": TrainingInput(...) } ) ``` 作业前缀默认情况下关闭。要选择使用此特征，请使用 `PipelineDefinitionConfig` 的 `use_custom_job_prefix` 选项，如以下代码片段所示： ``` from sagemaker.workflow.pipeline_definition_config import PipelineDefinitionConfig # Create a definition configuration and toggle on custom prefixing definition_config = PipelineDefinitionConfig(use_custom_job_prefix=True); # Create a pipeline with a custom prefix pipeline = Pipeline( name="MyJobPrefixedPipeline", parameters=[...] steps=[...] pipeline_definition_config=definition_config ) ``` 创建并运行管道。以下示例创建并运行管道，还演示了如何关闭作业前缀并重新运行管道。 ``` pipeline.create(role_arn=sagemaker.get_execution_role()) # Optionally, call definition() to confirm your prefixed job names are in the built JSON pipeline.definition() pipeline.start() # To run a pipeline without custom-prefixes, toggle off use_custom_job_prefix, update the pipeline # via upsert() or update(), and start a new run definition_config = PipelineDefinitionConfig(use_custom_job_prefix=False) pipeline.pipeline_definition_config = definition_config pipeline.update() execution = pipeline.start() ``` 同样，您可以为现有管道开启该特征，然后启动使用作业前缀的新运行。 ``` definition_config = PipelineDefinitionConfig(use_custom_job_prefix=True) pipeline.pipeline_definition_config = definition_config pipeline.update() execution = pipeline.start() ``` 最后，您可以在管道执行时调用 `list_steps` 来查看自定义前缀的作业。 ``` steps = execution.list_steps() prefixed_training_job_name = steps['PipelineExecutionSteps'][0]['Metadata']['TrainingJob']['Arn'] ``` ## 自定义对管道版本的访问权限 您可以使用`sagemaker:PipelineVersionId`条件键授予对特定版本的 Amazon Pi SageMaker pelines 的自定义访问权限。例如，以下策略仅允许对版本 ID 为 6 及以上的管道启动执行或更新管道版本。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": { "Sid": "AllowStartPipelineExecution", "Effect": "Allow", "Action": [ "sagemaker:StartPipelineExecution", "sagemaker:UpdatePipelineVersion" ], "Resource": "*", "Condition": { "NumericGreaterThanEquals": { "sagemaker:PipelineVersionId": 6 } } } } ``` ------ 有关支持的条件键的更多信息，请参阅 [Amazon A SageMaker I 的条件键](https://docs.amazonaws.cn//service-authorization/latest/reference/list_amazonsagemaker.html#amazonsagemaker-policy-keys)。 ## 使用管道的服务控制策略 服务控制策略 (SCPs) 是一种组织策略，可用于管理组织中的权限。 SCPs 集中控制组织中所有账户的最大可用权限。通过在组织中使用 Pipelines，您可以确保数据科学家无需与 Amazon 控制台交互即可管理您的管道执行。  如果您的 SCP 使用的 VPC 限制访问 Amazon S3，则需要采取措施允许您的管道访问其他 Amazon S3 资源。 要允许 Pipelines 使用 `JsonGet` 功能访问 VPC 外部的 Amazon S3，请更新组织的 SCP，确保使用 Pipelines 的角色可以访问 Amazon S3。为此，请使用主体标签和条件键为管道执行器通过管道执行角色使用的角色创建一个异常情况。 **允许 Pipelines 访问 VPC 以外的 Amazon S3** 1. 按照[标记 IAM 用户和角色](https://docs.amazonaws.cn/IAM/latest/UserGuide/id_tags.html)中的步骤为管道执行角色创建唯一标签。 1. 使用您创建的标签的 `Aws:PrincipalTag IAM` 条件键，在您的 SCP 中授予一个例外。有关更多信息，请参阅[创建、更新和删除服务控制策略](https://docs.amazonaws.cn/organizations/latest/userguide/orgs_manage_policies_scps_create.html)。 # 为 Pipelines 设置跨账户支持 建立跨账户支持 Amazon Pipel SageMaker ines 的跨账户支持使您能够与使用不同 Amazon 账户运营的其他团队或组织在机器学习管道上进行协作。通过设置跨账户管道共享，您可以对管道授予受控访问权限，允许其他账户查看管道详情、触发执行和监控运行。以下主题介绍如何设置跨账户管道共享、共享资源可用的不同权限策略，以及如何通过对 SageMaker AI 的直接 API 调用来访问共享管道实体并与之交互。 ## 设置跨账户管道共享 设置跨账户管道共享 SageMaker AI 使用 R [es Amazon ource Access Manager](https://docs.amazonaws.cn/ram/latest/userguide/what-is.html) (Amazon RAM) 来帮助你安全地跨账户共享管道实体。 ### 创建资源共享 1. 通过 [Amazon RAM 控制台](https://console.amazonaws.cn/ram/home)选择**创建资源共享**。 1. 指定资源共享详细信息时，请选择管道资源类型，并选择要共享的一个或多个管道。当您与任何其他账户共享管道时，其所有执行也会被隐式共享。 1. 将权限与资源共享关联。选择默认只读权限策略或扩展的管道执行权限策略。有关更多详细信息，请参阅 [Pipelines 资源的权限策略](#build-and-manage-xaccount-permissions)。 **注意** 如果您选择扩展管道执行策略，请注意共享账户调用的任何启动、停止和重试命令都使用共享管道的 Amazon 账户中的资源。 1. 使用 Amazon 账户指定 IDs 要向其授予共享资源访问权限的账户。 1. 查看您的资源共享配置，然后选择**创建资源共享**。可能需要几分钟时间来完成资源共享和主体关联。 有关更多信息，请参阅《Amazon Resource Access Manager 用户指南》**中的[共享 Amazon 资源](https://docs.amazonaws.cn/ram/latest/userguide/getting-started-sharing.html)。 ### 获取对资源共享邀请的回复 设置资源共享和主体关联后，指定的 Amazon 账户会收到加入资源共享的邀请。 Amazon 账户必须接受邀请才能访问任何共享资源。 有关通过接受资源共享邀请的更多信息 Amazon RAM，请参阅 [Resource Acc *ess Manager 用户指南*中的使用共享 Amazon 资源](https://docs.amazonaws.cn/ram/latest/userguide/getting-started-shared.html)。Amazon ## Pipelines 资源的权限策略 创建资源共享时，请选择两个支持的权限策略之一，将其与 SageMaker AI 管道资源类型相关联。这两个策略都授予对任何选定管道及其所有执行的访问权限。 ### 默认只读权限 `AWSRAMDefaultPermissionSageMakerPipeline` 策略允许执行以下只读操作： ``` "sagemaker:DescribePipeline" "sagemaker:DescribePipelineDefinitionForExecution" "sagemaker:DescribePipelineExecution" "sagemaker:ListPipelineExecutions" "sagemaker:ListPipelineExecutionSteps" "sagemaker:ListPipelineParametersForExecution" "sagemaker:Search" ``` ### 扩展的管道执行权限 `AWSRAMPermissionSageMakerPipelineAllowExecution` 策略包括默认策略中的所有只读权限，还允许共享账户启动、停止和重试管道执行。 **注意** 使用扩展管道执行权限策略时，请注意 Amazon 资源使用情况。使用此策略时，允许共享账户启动、停止和重试管道执行。用于共享管道执行的所有资源均由拥有者账户使用。 扩展的管道执行权限策略允许执行以下操作： ``` "sagemaker:DescribePipeline" "sagemaker:DescribePipelineDefinitionForExecution" "sagemaker:DescribePipelineExecution" "sagemaker:ListPipelineExecutions" "sagemaker:ListPipelineExecutionSteps" "sagemaker:ListPipelineParametersForExecution" "sagemaker:StartPipelineExecution" "sagemaker:StopPipelineExecution" "sagemaker:RetryPipelineExecution" "sagemaker:Search" ``` ## 通过直接 API 调用访问共享管道实体 跨账户渠道共享设置完成后，您可以使用管道 ARN 调用以下 SageMaker API 操作： **注意** 只有当 API 命令包含在与资源共享关联的权限中时，才能调用这些命令。如果您选择`AWSRAMPermissionSageMakerPipelineAllowExecution`策略，则启动、停止和重试命令将使用共享管道的 Amazon 账户中的资源。 + [DescribePipeline](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_DescribePipeline.html) + [DescribePipelineDefinitionForExecution](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_DescribePipelineDefinitionForExecution.html) + [DescribePipelineExecution](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_DescribePipelineExecution.html) + [ListPipelineExecutions](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_ListPipelineExecutions.html) + [ListPipelineExecutionSteps](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_ListPipelineExecutionSteps.html) + [ListPipelineParametersForExecution](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_ListPipelineParametersForExecution.html) + [StartPipelineExecution](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_StartPipelineExecution.html) + [StopPipelineExecution](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_StopPipelineExecution.html) + [RetryPipelineExecution](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_RetryPipelineExecution.html) # Pipeline 参数 您可以使用参数在管道定义中引入变量。您可以在整个管道定义过程中引用您定义的参数。参数具有默认值，您可以在开始执行管道时通过指定参数值来覆盖该值。默认值必须是与参数类型匹配的实例。步骤定义中使用的所有参数都必须在管道定义中定义。本主题介绍可以定义的参数以及如何执行这些参数。 Amaz SageMaker on Pipelines 支持以下参数类型： + `ParameterString` - 表示字符串参数。 + `ParameterInteger` - 表示整数参数。 + `ParameterFloat` - 表示浮点参数。 + `ParameterBoolean` - 表示布尔 Python 类型。 参数采用以下格式： ``` = ( name="", default_value= ) ``` 下面的示例显示了一个示例参数实施。 ``` from sagemaker.workflow.parameters import ( ParameterInteger, ParameterString, ParameterFloat, ParameterBoolean ) processing_instance_count = ParameterInteger( name="ProcessingInstanceCount", default_value=1 ) ``` 创建管道时传递参数，如以下示例所示。 ``` pipeline = Pipeline( name=pipeline_name, parameters=[ processing_instance_count ], steps=[step_process] ) ``` 还可将不同于默认值的参数值传递给管道执行，如以下示例所示。 ``` execution = pipeline.start( parameters=dict( ProcessingInstanceCount="2", ModelApprovalStatus="Approved" ) ) ``` 你可以使用 SageMaker Python SDK 函数来操作参数，比如`[ sagemaker.workflow.functions.Join](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.functions.Join)`。有关参数的更多信息，请参阅[ SageMaker 管道参数](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#parameters)。 有关管道参数的已知限制，请参阅 Amaz [on Pyth SageMaker on](https://sagemaker.readthedocs.io/en/stable) SD *[K 中的限制-参数化](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#parameterization)*。 # Pipelines 步骤 Pipelines 步骤 Pipelines 由多个步骤组成。这些步骤使用属性定义管道采取的操作以及各步骤之间的关系。下一页将介绍步骤的类型、属性以及它们之间的关系。 **Topics** + [ # 增加一个步骤 ](build-and-manage-steps-types.md) + [ # 添加集成 ](build-and-manage-steps-integration.md) + [ ## 步骤属性 ](#build-and-manage-properties) + [ ## 步骤并行 ](#build-and-manage-parallelism) + [ ## 步骤之间的数据依赖性 ](#build-and-manage-data-dependency) + [ ## 步骤之间的自定义依赖关系 ](#build-and-manage-custom-dependency) + [ ## 自定义映像一步到位 ](#build-and-manage-images) # 增加一个步骤 下文介绍了每种步骤类型的要求，并提供了步骤的实现示例，以及如何将步骤添加到 Pipelines 中。这些都不是可行的实施方案，因为它们没有提供所需的资源和投入。有关实施这些步骤的教程，请参阅[Pipelines 操作](pipelines-build.md)。 **注意** 您还可以使用 `@step` 装饰器将本地机器学习代码转换为 Pipelines 步骤，从而从本地机器学习代码中创建步骤。有关更多信息，请参阅 [@step decorator](#step-type-custom)。 Amaz SageMaker on Pipelines 支持以下步骤类型： + [执行代码](#step-type-executecode) [Processing](#step-type-processing) + [训练](#step-type-training) + [优化](#step-type-tuning) + [AutoML](#step-type-automl) + [`Model`](#step-type-model) + [`Create model`](#step-type-create-model) + [`Register model`](#step-type-register-model) + [`Deploy model (endpoint)`](#step-type-deploy-model-endpoint) + [转换](#step-type-transform) + [条件](#step-type-condition) + [`Callback`](#step-type-callback) + [Lambda](#step-type-lambda) + [`ClarifyCheck`](#step-type-clarify-check) + [`QualityCheck`](#step-type-quality-check) + [EMR](#step-type-emr) + [笔记本作业](#step-type-notebook-job) + [Fail](#step-type-fail) ## @step decorator 如果您想在 Pipelines 用户界面中编排利用高级 SageMaker AI 功能或其他 Amazon 服务的自定义 ML 作业，请使用。 drag-and-drop [执行代码步骤](#step-type-executecode) 您可以使用 `@step` 装饰器从本地机器学习代码中创建一个步骤。测试代码后，您可以通过使用装饰器对其进行注释来将该函数转换为 SageMaker AI 管道步骤。`@step`当您将 `@step` 装饰功能的输出作为一个步骤传递给管道时，Pipelines 会创建并运行管道。您还可以创建一个多步骤 DAG 管道，其中包括一个或多个`@step`经过装饰的函数以及传统的 SageMaker AI 工作流步骤。有关如何使用 `@step` 装饰器创建步骤的更多详情，请参阅 [Lift-and-shift 使用 @step 装饰器的 Python 代码](pipelines-step-decorator.md)。 ## 执行代码步骤 执行代码 在 Pipelines drag-and-drop UI 中，您可以使用 “**执行代码**” 步骤将自己的代码作为工作流步骤运行。您可以上传一个 Python 函数、脚本或笔记本，作为管道的一部分执行。如果您想编排利用高级 SageMaker AI 功能或其他 Amazon 服务的自定义 ML 作业，则应使用此步骤。 “**执行代码**” 步骤将文件上传到你的 Amazon A SageMaker I 默认 Amazon S3 存储桶。该存储桶可能未设置所需的跨源资源共享 (CORS) 权限。要了解有关配置 CORS 权限的更多信息，请参阅 [输入映像数据的 CORS 要求](sms-cors-update.md)。 “**执行代码”** 步骤使用 Amazon SageMaker 训练作业来运行您的代码。确保您的 IAM 角色具有 `sagemaker:DescribeTrainingJob` 和 `sagemaker:CreateTrainingJob` API 权限。要详细了解 Amazon A SageMaker I 的所有必需权限以及如何设置这些权限，请参阅[Amazon SageMaker AI API 权限：操作、权限和资源参考](api-permissions-reference.md)。 要使用管道设计器向管道添加执行代码步骤，请执行以下操作： 1. 按照中的说明打开 Amazon SageMaker Studio 控制台[启动亚马逊 SageMaker Studio](studio-updated-launch.md)。 1. 在左侧导航窗格中，选择 **Pipelines**。 1. 选择**创建**。 1. 选择**空白**。 1. 在左侧边栏中选择**执行代码**，然后将其拖到画布上。 1. 在画布中，选择添加的**执行代码**步骤。 1. 在右侧边栏中，填写**设置**和**详情**标签中的表格。 1. 您可以上传单个文件来执行，也可以上传包含多个构件的压缩文件夹。 1. 对于单个文件上传，您可以为笔记本、python 函数或脚本提供可选参数。 1. 在提供 Python 函数时，必须以 `file.py:` 格式提供处理程序。 1. 对于上传压缩文件夹，必须提供代码的相对路径，您还可以选择提供压缩文件夹内 `requirements.txt` 文件或初始化脚本的路径。 1. 如果画布上有任何步骤紧接在您添加的**执行代码**步骤之前，请单击并拖动光标从该步骤到**执行代码**步骤，以创建边缘。 1. 如果画布中包含任何紧接您添加的**执行代码**步骤的步骤，请单击并拖动光标从**执行代码**步骤到该步骤，以创建边缘。可以引用**执行代码**步骤的输出作为 Python 函数。 ## 处理步骤 Processing 使用处理步骤创建用于数据处理的处理作业。有关处理作业的更多信息，请参阅[处理数据和评估模型](https://docs.amazonaws.cn/sagemaker/latest/dg/processing-job.html)。 ------ #### [ Pipeline Designer ] 要使用管道设计器向管道添加处理步骤，请执行以下操作： 1. 按照中的说明打开 Amazon SageMaker Studio 控制台[启动亚马逊 SageMaker Studio](studio-updated-launch.md)。 1. 在左侧导航窗格中，选择 **Pipelines**。 1. 选择**创建**。 1. 在左侧边栏中选择**处理数据**，然后将其拖到画布上。 1. 在画布中，选择添加的**处理数据**步骤。 1. 在右侧边栏中，填写**设置**和**详情**标签中的表格。有关这些选项卡中字段的信息，请参阅 [sagemaker.workflow.steps。 ProcessingStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.ProcessingStep)。 1. 如果画布上有任何步骤紧接在您添加的**处理数据**步骤之前，请单击并拖动光标从该步骤到**处理数据**步骤，以创建边缘。 1. 如果画布中包含任何紧接您添加的**处理数据**步骤的步骤，请单击并拖动光标从**处理数据**步骤到该步骤，以创建边缘。 ------ #### [ SageMaker Python SDK ] 处理步骤需要处理器、定义处理代码的 Python 脚本、用于处理的输出以及作业参数。以下示例说明如何创建 `ProcessingStep` 定义。 ``` from sagemaker.sklearn.processing import SKLearnProcessor sklearn_processor = SKLearnProcessor(framework_version='1.0-1', role=, instance_type='ml.m5.xlarge', instance_count=1) ``` ``` from sagemaker.processing import ProcessingInput, ProcessingOutput from sagemaker.workflow.steps import ProcessingStep inputs = [ ProcessingInput(source=, destination="/opt/ml/processing/input"), ] outputs = [ ProcessingOutput(output_name="train", source="/opt/ml/processing/train"), ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"), ProcessingOutput(output_name="test", source="/opt/ml/processing/test") ] step_process = ProcessingStep( name="AbaloneProcess", step_args = sklearn_processor.run(inputs=inputs, outputs=outputs, code="abalone/preprocessing.py") ) ``` **传递运行时参数** 以下示例说明如何将运行时参数从 PySpark处理器传递给`ProcessingStep`。 ``` from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.spark.processing import PySparkProcessor from sagemaker.processing import ProcessingInput, ProcessingOutput from sagemaker.workflow.steps import ProcessingStep pipeline_session = PipelineSession() pyspark_processor = PySparkProcessor( framework_version='2.4', role=, instance_type='ml.m5.xlarge', instance_count=1, sagemaker_session=pipeline_session, ) step_args = pyspark_processor.run( inputs=[ProcessingInput(source=, destination="/opt/ml/processing/input"),], outputs=[ ProcessingOutput(output_name="train", source="/opt/ml/processing/train"), ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"), ProcessingOutput(output_name="test", source="/opt/ml/processing/test") ], code="preprocess.py", arguments=None, ) step_process = ProcessingStep( name="AbaloneProcess", step_args=step_args, ) ``` 有关处理步骤要求的更多信息，请参阅 [sagemaker.workflow.steps。 ProcessingStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.ProcessingStep)文档。有关深入的示例，请参阅使用 [Amazon Pipelines 编排任务以训练和评估模型的 SageMaker 示例](https://github.com/aws/amazon-sagemaker-examples/blob/62de6a1fca74c7e70089d77e36f1356033adbe5f/sagemaker-pipelines/tabular/abalone_build_train_deploy/sagemaker-pipelines-preprocess-train-evaluate-batch-transform.ipynb)笔记本。*为特征工程定义处理步骤*部分包含更多信息。 ------ ## 训练步骤 训练 您可以使用训练步骤创建训练作业来训练模型。有关训练作业的更多信息，请参阅[使用 Amazon A SageMaker I 训练模型](https://docs.amazonaws.cn/sagemaker/latest/dg/how-it-works-training.html)。 训练步骤需要估算器以及训练和验证数据输入。 ------ #### [ Pipeline Designer ] 要使用管道设计器向管道添加训练步骤，请执行以下操作： 1. 按照中的说明打开 Amazon SageMaker Studio 控制台[启动亚马逊 SageMaker Studio](studio-updated-launch.md)。 1. 在左侧导航窗格中，选择 **Pipelines**。 1. 选择**创建**。 1. 选择**空白**。 1. 在左侧边栏中选择**训练模型**，然后将其拖到画布上。 1. 在画布中选择添加的**训练模型**步骤。 1. 在右侧边栏中，填写**设置**和**详情**标签中的表格。有关这些选项卡中字段的信息，请参阅 [sagemaker.workflow.steps。 TrainingStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.TrainingStep)。 1. 如果画布上有任何步骤紧接在您添加的**训练模型**步骤之前，请单击并拖动光标从该步骤到**训练模型**步骤，以创建边缘。 1. 如果画布上有任何步骤紧接着您添加的**训练模型**步骤，请单击并拖动光标从**训练模型**步骤到该步骤，以创建边缘。 ------ #### [ SageMaker Python SDK ] 以下示例说明如何创建 `TrainingStep` 定义。有关训练步骤要求的更多信息，请参阅 [sagemaker.workflow.steps。 TrainingStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.TrainingStep)文档。 ``` from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.inputs import TrainingInput from sagemaker.workflow.steps import TrainingStep from sagemaker.xgboost.estimator import XGBoost pipeline_session = PipelineSession() xgb_estimator = XGBoost(..., sagemaker_session=pipeline_session) step_args = xgb_estimator.fit( inputs={ "train": TrainingInput( s3_data=step_process.properties.ProcessingOutputConfig.Outputs[ "train" ].S3Output.S3Uri, content_type="text/csv" ), "validation": TrainingInput( s3_data=step_process.properties.ProcessingOutputConfig.Outputs[ "validation" ].S3Output.S3Uri, content_type="text/csv" ) } ) step_train = TrainingStep( name="TrainAbaloneModel", step_args=step_args, ) ``` ------ ## 优化步骤 优化 您可以使用优化步骤来创建超参数优化作业，也称为超参数优化 (HPO)。超参数调整作业运行多个训练作业，每个作业产生一个模型版本。有关超参数优化的更多信息，请参阅[使用 SageMaker AI 自动调整模型](automatic-model-tuning.md)。 调优作业与管道的 SageMaker AI 实验相关联，而训练作业是作为试验创建的。有关更多信息，请参阅 [实验集成](pipelines-experiments.md)。 调整步骤需要[HyperparameterTuner](https://sagemaker.readthedocs.io/en/stable/api/training/tuner.html)和训练输入。您可以通过指定 `HyperparameterTuner` 的 `warm_start_config` 参数来再训练以前的优化作业。有关超参数优化和温启动的更多信息，请参阅[运行热启动超参数调优作业](automatic-model-tuning-warm-start.md)。 [你可以使用 sagemaker.workflow.step [s 的 get\$1top\$1model\$1s3\$1uri](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.TuningStep.get_top_model_s3_uri) 方法。 TuningStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.TuningStep)class，用于从性能最好的模型版本之一中获取模型工件。有关显示如何在 SageMaker AI 管道中使用调整步骤的笔记本，请参阅 [sagemaker-pipelines-tuning-step.ipynb。](https://github.com/aws/amazon-sagemaker-examples/blob/main/sagemaker-pipelines/tabular/tuning-step/sagemaker-pipelines-tuning-step.ipynb) **重要** Amaz SageMaker on Python SDK v2.48.0 和 Amazon Studio Classic v3.8.0 中引入了调整步骤。 SageMaker 使用调整步骤前必须更新 Studio Classic，否则不会显示管道 DAG。要更新 Studio Classic，请参阅 [关闭并更新 Amazon SageMaker Studio 经典版](studio-tasks-update-studio.md)。 以下示例说明如何创建 `TuningStep` 定义。 ``` from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.tuner import HyperparameterTuner from sagemaker.inputs import TrainingInput from sagemaker.workflow.steps import TuningStep tuner = HyperparameterTuner(..., sagemaker_session=PipelineSession()) step_tuning = TuningStep( name = "HPTuning", step_args = tuner.fit(inputs=TrainingInput(s3_data="s3://amzn-s3-demo-bucket/my-data")) ) ``` **获取最佳模型版本** 以下示例说明如何使用 `get_top_model_s3_uri` 方法从优化作业中获取最佳模型版本。最多只能根据排名前50位的版本进行排名[HyperParameterTuningJobObjective](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_HyperParameterTuningJobObjective.html)。`top_k` 参数是版本的索引，其中 `top_k=0` 是性能最好的版本，`top_k=49` 是性能最差的版本。 ``` best_model = Model( image_uri=image_uri, model_data=step_tuning.get_top_model_s3_uri( top_k=0, s3_bucket=sagemaker_session.default_bucket() ), ... ) ``` 有关调整步骤要求的更多信息，请参阅 [sagemaker.workflow.steps。 TuningStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.TuningStep)文档。 ## 微调步骤 微调 Fine-tuning 在新数据集上训练来自 Amaz SageMaker JumpStart on 的预训练基础模型。这个过程也称为转移学习，可以使用较小数据集和较短的训练时间生成准确模型。对模型进行微调时，可以使用默认数据集，也可以选择自己的数据。要了解有关微调基础模型的更多信息 JumpStart，请参阅[微调模型](jumpstart-fine-tune.md)。 微调步骤使用 Amazon SageMaker 训练作业自定义您的模型。确保您的 IAM 角色具有 `sagemaker:DescribeTrainingJob` 和 `sagemaker:CreateTrainingJob` API 权限，以便在管道中执行微调作业。API 权限，以便在管道中执行微调作业。要了解有关 Amazon A SageMaker I 所需权限以及如何设置权限的更多信息，请参阅[Amazon SageMaker AI API 权限：操作、权限和资源参考](api-permissions-reference.md)。 要使用 drag-and-drop编辑器向管道中添加 Fine-**Tune 模型**步骤，请执行以下步骤： 1. 按照 [启动亚马逊 SageMaker Studio](studio-updated-launch.md) 中的说明打开 Studio 管理控制台。 1. 在左侧导航窗格中，选择 **Pipelines**。 1. 选择**创建**。 1. 选择**空白**。 1. 在左侧边栏中选择**微调模型**，然后将其拖动到画布上。 1. 在画布中，选择添加的**微调模型**步骤。 1. 在右侧边栏中，填写**设置**和**详情**标签中的表格。 1. 如果画布上有任何步骤紧接在您添加的**微调模型**步骤之前，请单击并拖动光标从该步骤到**微调模型**步骤，以创建边缘。 1. 如果画布上有任何步骤紧接着您添加的**微调模型**步骤，请单击并拖动光标从**微调模型**步骤到该步骤，以创建边缘。 ## AutoML 步骤 AutoML 使用 [AutoML](https://sagemaker.readthedocs.io/en/stable/api/training/automl.html) API 创建 AutoML 作业以自动训练模型。有关 AutoML 任务的更多信息，请参阅[使用 Ama SageMaker zon Autopilot 自动开发模型](https://docs.amazonaws.cn/sagemaker/latest/dg/autopilot-automate-model-development.html)。 **注意** 目前，AutoML 步骤仅支持[组合训练模式](https://docs.amazonaws.cn/sagemaker/latest/dg/autopilot-model-support-validation.html)。 以下示例说明如何使用 `AutoMLStep` 创建定义。 ``` from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.workflow.automl_step import AutoMLStep pipeline_session = PipelineSession() auto_ml = AutoML(..., role="", target_attribute_name="my_target_attribute_name", mode="ENSEMBLING", sagemaker_session=pipeline_session) input_training = AutoMLInput( inputs="s3://amzn-s3-demo-bucket/my-training-data", target_attribute_name="my_target_attribute_name", channel_type="training", ) input_validation = AutoMLInput( inputs="s3://amzn-s3-demo-bucket/my-validation-data", target_attribute_name="my_target_attribute_name", channel_type="validation", ) step_args = auto_ml.fit( inputs=[input_training, input_validation] ) step_automl = AutoMLStep( name="AutoMLStep", step_args=step_args, ) ``` **获取最佳模型版本** AutoML 步骤会自动训练多个候选模型。使用 `get_best_auto_ml_model` 方法从 AutoML 作业中获取目标指标最佳的模型，如下所示。您还必须使用 IAM `role` 访问模型构件。 ``` best_model = step_automl.get_best_auto_ml_model(role=) ``` 有关更多信息，请参阅 Pyth SageMaker on 软件开发工具包中的 [AutoML](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.automl_step.AutoMLStep) 步骤。 ## 模型步骤 `Model` `ModelStep`使用创建或注册 A SageMaker I 模型。有关`ModelStep`要求的更多信息，请参阅 sagemaker.workflow. [model\$1step。 ModelStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.model_step.ModelStep)文档。 ### 创建模型 您可以使用创建 `ModelStep` A SageMaker I 模型。A `ModelStep` 需要模型工件和有关创建模型所需的 SageMaker AI 实例类型的信息。有关 SageMaker AI 模型的更多信息，请参阅使用 [Amazon A SageMaker I 训练模型](https://docs.amazonaws.cn/sagemaker/latest/dg/how-it-works-training.html)。 以下示例说明如何创建 `ModelStep` 定义。 ``` from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.model import Model from sagemaker.workflow.model_step import ModelStep step_train = TrainingStep(...) model = Model( image_uri=pytorch_estimator.training_image_uri(), model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts, sagemaker_session=PipelineSession(), role=role, ) step_model_create = ModelStep( name="MyModelCreationStep", step_args=model.create(instance_type="ml.m5.xlarge"), ) ``` ### 注册模型 您可以使用 a `ModelStep` 在 Amazon SageMaker 模型注册中注册`sagemaker.model.Model`或。`sagemaker.pipeline.PipelineModel``PipelineModel` 表示推理管道，是一个由处理推理请求的容器线性序列组成的模型。有关如何注册模型的更多信息，请参阅[利用模型注册中心进行模型注册部署](model-registry.md)。 以下示例说明如何创建一个注册 `PipelineModel` 的 `ModelStep`。 ``` import time from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.sklearn import SKLearnModel from sagemaker.xgboost import XGBoostModel pipeline_session = PipelineSession() code_location = 's3://{0}/{1}/code'.format(bucket_name, prefix) sklearn_model = SKLearnModel( model_data=processing_step.properties.ProcessingOutputConfig.Outputs['model'].S3Output.S3Uri, entry_point='inference.py', source_dir='sklearn_source_dir/', code_location=code_location, framework_version='1.0-1', role=role, sagemaker_session=pipeline_session, py_version='py3' ) xgboost_model = XGBoostModel( model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts, entry_point='inference.py', source_dir='xgboost_source_dir/', code_location=code_location, framework_version='0.90-2', py_version='py3', sagemaker_session=pipeline_session, role=role ) from sagemaker.workflow.model_step import ModelStep from sagemaker import PipelineModel pipeline_model = PipelineModel( models=[sklearn_model, xgboost_model], role=role,sagemaker_session=pipeline_session, ) register_model_step_args = pipeline_model.register( content_types=["application/json"], response_types=["application/json"], inference_instances=["ml.t2.medium", "ml.m5.xlarge"], transform_instances=["ml.m5.xlarge"], model_package_group_name='sipgroup', ) step_model_registration = ModelStep( name="AbaloneRegisterModel", step_args=register_model_step_args, ) ``` ## 创建模型步骤 `Create model` 您可以使用 “创建模型” 步骤来创建 A SageMaker I 模型。有关 SageMaker AI 模型的更多信息，请参阅[使用 Amazon 训练模型 SageMaker](how-it-works-training.md)。 创建模型步骤需要模型工件和有关创建模型所需的 SageMaker AI 实例类型的信息。下面的示例展示了如何创建创建模型步骤定义。有关创建模型步骤要求的更多信息，请参阅 [sagemaker.workflow.steps。 CreateModelStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.CreateModelStep)文档。 ------ #### [ Pipeline Designer ] 要在管道中添加创建模型步骤，请执行以下操作： 1. 按照 [启动亚马逊 SageMaker Studio](studio-updated-launch.md) 中的说明打开 Studio 管理控制台。 1. 在左侧导航窗格中，选择 **Pipelines**。 1. 选择**创建**。 1. 选择**空白**。 1. 在左侧边栏中选择**创建模型**，然后将其拖到画布上。 1. 在画布中，选择添加的**创建模型**步骤。 1. 在右侧边栏中，填写**设置**和**详情**标签中的表格。有关这些选项卡中字段的信息，请参阅 [sagemaker.workflow.steps。 CreateModelStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.CreateModelStep)。 1. 如果画布上有任何步骤紧接在您添加的**创建模型**步骤之前，请单击并拖动光标从该步骤到**创建模型**步骤，以创建边缘。 1. 如果画布上有任何步骤紧接着您添加的**创建模型**步骤，请单击并拖动光标从**创建模型**步骤到该步骤，以创建边缘。 ------ #### [ SageMaker Python SDK ] **重要** 从 AI SageMaker Py [模型步骤](#step-type-model) thon SDK 版本 2.90.0 起，我们建议使用创建模型。 `CreateModelStep`将继续在以前版本的 SageMaker Python SDK 中运行，但不再受支持。 ``` from sagemaker.workflow.steps import CreateModelStep step_create_model = CreateModelStep( name="AbaloneCreateModel", model=best_model, inputs=inputs ) ``` ------ ## 注册模型步骤 `Register model` “注册模型” 步骤将模型注册到 “ SageMaker 模型注册表”。 ------ #### [ Pipeline Designer ] 要使用管道设计器从管道中注册模型，请执行以下操作： 1. 按照中的说明打开 Amazon SageMaker Studio 控制台[启动亚马逊 SageMaker Studio](studio-updated-launch.md)。 1. 在左侧导航窗格中，选择 **Pipelines**。 1. 选择**创建**。 1. 选择**空白**。 1. 在左侧边栏中选择**注册模型**，然后将其拖到画布上。 1. 在画布中，选择添加的**注册模型**步骤。 1. 在右侧边栏中，填写**设置**和**详情**标签中的表格。有关这些选项卡中字段的信息，请参阅 [sagemaker.workflow.step\$1collections。 RegisterModel](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.step_collections.RegisterModel)。 1. 如果画布上有任何步骤紧接在您添加的**注册模型**步骤之前，请单击并拖动光标从该步骤到**注册模型**步骤，以创建边缘。 1. 如果画布中包含任何紧接您添加的**注册模型**步骤的步骤，请单击并拖动光标从**注册模型**步骤到该步骤，以创建边缘。 ------ #### [ SageMaker Python SDK ] **重要** 我们建议使用[模型步骤](#step-type-model)从 AI SageMaker Python SDK 版本 2.90.0 起注册模型。 `RegisterModel`将继续在以前版本的 SageMaker Python SDK 中运行，但不再受支持。 [你可以使用`RegisterModel`步骤注册 sagemaker.model.model 或 [sagemaker.pipeline。](https://sagemaker.readthedocs.io/en/stable/api/inference/model.html) PipelineModel](https://sagemaker.readthedocs.io/en/stable/api/inference/pipeline.html#pipelinemodel)使用 Amazon SageMaker 模型注册表。`PipelineModel` 表示推理管道，是一个由处理推理请求的容器线性序列组成的模型。 有关如何注册模型的更多信息，请参阅[利用模型注册中心进行模型注册部署](model-registry.md)。有关`RegisterModel`步骤要求的更多信息，请参阅 s [agemaker.workflow.step\$1collections。 RegisterModel](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.step_collections.RegisterModel)文档。 以下示例说明如何创建注册 `PipelineModel` 的 `RegisterModel` 步骤。 ``` import time from sagemaker.sklearn import SKLearnModel from sagemaker.xgboost import XGBoostModel code_location = 's3://{0}/{1}/code'.format(bucket_name, prefix) sklearn_model = SKLearnModel(model_data=processing_step.properties.ProcessingOutputConfig.Outputs['model'].S3Output.S3Uri, entry_point='inference.py', source_dir='sklearn_source_dir/', code_location=code_location, framework_version='1.0-1', role=role, sagemaker_session=sagemaker_session, py_version='py3') xgboost_model = XGBoostModel(model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts, entry_point='inference.py', source_dir='xgboost_source_dir/', code_location=code_location, framework_version='0.90-2', py_version='py3', sagemaker_session=sagemaker_session, role=role) from sagemaker.workflow.step_collections import RegisterModel from sagemaker import PipelineModel pipeline_model = PipelineModel(models=[sklearn_model,xgboost_model],role=role,sagemaker_session=sagemaker_session) step_register = RegisterModel( name="AbaloneRegisterModel", model=pipeline_model, content_types=["application/json"], response_types=["application/json"], inference_instances=["ml.t2.medium", "ml.m5.xlarge"], transform_instances=["ml.m5.xlarge"], model_package_group_name='sipgroup', ) ``` 如果未提供 `model`，则注册模型步骤需要使用估算器，如以下示例所示。 ``` from sagemaker.workflow.step_collections import RegisterModel step_register = RegisterModel( name="AbaloneRegisterModel", estimator=xgb_train, model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts, content_types=["text/csv"], response_types=["text/csv"], inference_instances=["ml.t2.medium", "ml.m5.xlarge"], transform_instances=["ml.m5.xlarge"], model_package_group_name=model_package_group_name, approval_status=model_approval_status, model_metrics=model_metrics ) ``` ------ ## 部署模型（端点）步骤 `Deploy model (endpoint)` 在管道设计器中，使用部署模型（端点）步骤将模型部署到端点。您可以创建一个新端点或使用现有端点。实时推理非常适合有实时、交互式、低延迟要求的推理工作负载。您可以将模型部署到 SageMaker AI Hosting 服务，并获得可用于推理的实时终端节点。这些端点可完全托管，并支持自动扩缩。要了解有关 SageMaker AI 中实时推理的更多信息，请参阅[实时推理](realtime-endpoints.md)。 在管道中添加部署模型步骤之前，请确保您的 IAM 角色具有以下权限： + `sagemaker:CreateModel` + `sagemaker:CreateEndpointConfig` + `sagemaker:CreateEndpoint` + `sagemaker:UpdateEndpoint` + `sagemaker:DescribeModel` + `sagemaker:DescribeEndpointConfig` + `sagemaker:DescribeEndpoint` 要详细了解 SageMaker AI 所需的所有权限以及如何设置这些权限，请参阅[Amazon SageMaker AI API 权限：操作、权限和资源参考](api-permissions-reference.md)。 要在 drag-and-drop编辑器中向流水线添加模型部署步骤，请完成以下步骤： 1. 按照 [启动亚马逊 SageMaker Studio](studio-updated-launch.md) 中的说明打开 Studio 管理控制台。 1. 在左侧导航窗格中，选择 **Pipelines**。 1. 选择**创建**。 1. 选择**空白**。 1. 在左侧边栏中选择**部署模型（端点）**，然后将其拖到画布上。 1. 在画布中，选择添加的**部署模型（端点）**步骤。 1. 在右侧边栏中，填写**设置**和**详情**标签中的表格。 1. 如果画布上有任何步骤紧接在您添加的**部署模型（端点）**步骤之前，请单击并拖动光标，从该步骤到**部署模型（端点）**步骤，以创建边缘。 1. 如果画布中包含任何紧接您添加的**部署模型（端点）**步骤的步骤，请单击并将光标从**部署模型（端点）**步骤拖动到该步骤，以创建边缘。 ## 转换步骤 转换 您可以使用转换步骤进行批量转换，在整个数据集上运行推理。有关批量转换的更多信息，请参阅[利用推理管道进行批量转换](inference-pipeline-batch.md)。 转换步骤需要转换器和用于运行批量转换的数据。下面的示例显示了如何创建变换步骤定义。有关转换步骤要求的更多信息，请参阅 [sagemaker.workflow.steps。 TransformStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.TransformStep)文档。 ------ #### [ Pipeline Designer ] 要使用 drag-and-drop可视化编辑器向管道添加批量转换步骤，请执行以下操作： 1. 按照 [启动亚马逊 SageMaker Studio](studio-updated-launch.md) 中的说明打开 Studio 管理控制台。 1. 在左侧导航窗格中，选择 **Pipelines**。 1. 选择**创建**。 1. 选择**空白**。 1. 在左侧边栏中选择**部署模型（批量转换）**，然后将其拖到画布上。 1. 在画布中，选择添加的**部署模型（批量转换）**步骤。 1. 在右侧边栏中，填写**设置**和**详情**标签中的表格。有关这些选项卡中字段的信息，请参阅 [sagemaker.workflow.steps。 TransformStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.TransformStep)。 1. 如果画布上有任何步骤紧接在您添加的**部署模型（批量转换）**步骤之前，请单击并拖动光标从该步骤到**部署模型（批量转换）**步骤，以创建边缘。 1. 如果画布中包含任何紧接您添加的**部署模型（批量转换）**步骤的步骤，请单击并将光标从**部署模型（批量转换）**步骤拖动到该步骤，以创建边缘。 ------ #### [ SageMaker Python SDK ] ``` from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.transformer import Transformer from sagemaker.inputs import TransformInput from sagemaker.workflow.steps import TransformStep transformer = Transformer(..., sagemaker_session=PipelineSession()) step_transform = TransformStep( name="AbaloneTransform", step_args=transformer.transform(data="s3://amzn-s3-demo-bucket/my-data"), ) ``` ------ ## 条件步骤 条件 您可以使用条件步骤来评估步骤属性的条件，以评估管道中下一步应该采取的操作。 需要一个条件步骤： + 一个条件列表。 + 如果条件评估结果为 `true`，则要运行的步骤列表。 + 如果条件评估结果为 `false`，则要运行的步骤列表。 ------ #### [ Pipeline Designer ] 要使用 Pipeline Designer 向管道添加条件步骤，请执行以下操作： 1. 按照中的说明打开 Amazon SageMaker Studio 控制台[启动亚马逊 SageMaker Studio](studio-updated-launch.md)。 1. 在左侧导航窗格中，选择 **Pipelines**。 1. 选择**创建**。 1. 选择**空白**。 1. 在左侧边栏中选择**条件**，然后将其拖到画布上。 1. 在画布中，选择添加的**条件**步骤。 1. 在右侧边栏中，填写**设置**和**详情**标签中的表格。有关这些选项卡中字段的信息，请参阅 [sagemaker.workflow.condition\$1step。 ConditionStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.condition_step.ConditionStep)。 1. 如果画布上有任何步骤紧接在您添加的**条件**步骤之前，请单击并拖动光标从该步骤到**条件**步骤，以创建边缘。 1. 如果画布中包含任何紧接您添加的**条件**步骤的步骤，请单击并拖动光标从**条件**步骤到该步骤，以创建边缘。 ------ #### [ SageMaker Python SDK ] 以下示例说明如何创建 `ConditionStep` 定义。 **限制** + 管道不支持使用嵌套条件步骤。不能将一个条件步骤作为另一个条件步骤的输入进行传递。 + 一个条件步骤不能在两个分支中使用相同的步骤。如果需要在两个分支中使用相同的步骤功能，请复制该步骤并为其指定不同的名称。 ``` from sagemaker.workflow.conditions import ConditionLessThanOrEqualTo from sagemaker.workflow.condition_step import ConditionStep from sagemaker.workflow.functions import JsonGet cond_lte = ConditionLessThanOrEqualTo( left=JsonGet( step_name=step_eval.name, property_file=evaluation_report, json_path="regression_metrics.mse.value" ), right=6.0 ) step_cond = ConditionStep( name="AbaloneMSECond", conditions=[cond_lte], if_steps=[step_register, step_create_model, step_transform], else_steps=[] ) ``` 有关`ConditionStep`要求的更多信息，请参阅 sagemaker.workflow.con [dition\$1step。 ConditionStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#conditionstep)API 参考。有关支持条件的更多信息，请参阅 *[A SageMaker I Python SDK 文档中的 Amazon Pipelin SageMaker es-条件](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#conditions)*。 ------ ## 回调步骤 `Callback` 使用`Callback`步骤向您的工作流程中添加并非由 Amazon Pipelines 直接提供的其他流程和 Amazon SageMaker 服务。当 `Callback` 步骤运行时，会发生以下过程： + Pipelines 向客户指定的 Amazon Simple Queue Service (Amazon SQS) 队列发送消息。该信息包含一个 Pipelines 生成的标记和客户提供的输入参数列表。发送信息后，Pipelines 会等待客户的回复。 + 客户从 Amazon SQS 队列中检索消息并启动他们的自定义流程。 + 该过程完成后，客户调用以下方法之一 APIs 并提交 Pipelines 生成的令牌： + [SendPipelineExecutionStepSuccess](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_SendPipelineExecutionStepSuccess.html)，以及输出参数列表 + [SendPipelineExecutionStepFailure](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_SendPipelineExecutionStepFailure.html)，以及失败原因 + API 调用会导致 Pipelines 继续执行管道流程或使流程失败。 有关`Callback`步骤要求的更多信息，请参阅 sagemaker.workflow.cal [lback\$1step。 CallbackStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.callback_step.CallbackStep)文档。有关完整的解决方案，请参阅[使用回调步骤扩展 SageMaker 管道以包含自定义步骤](https://www.amazonaws.cn/blogs/machine-learning/extend-amazon-sagemaker-pipelines-to-include-custom-steps-using-callback-steps/)。 **重要** `Callback`亚马逊 SageMaker Python SDK v2.45.0 和 Amazon SageMaker Studio Classic v3.6.2 中引入了步骤。在使用 `Callback` 步骤之前，您必须更新 Studio Classic，否则不会显示管道 DAG。要更新 Studio Classic，请参阅 [关闭并更新 Amazon SageMaker Studio 经典版](studio-tasks-update-studio.md)。 下面的示例展示了上述程序的执行过程。 ``` from sagemaker.workflow.callback_step import CallbackStep step_callback = CallbackStep( name="MyCallbackStep", sqs_queue_url="https://sqs.us-east-2.amazonaws.com/012345678901/MyCallbackQueue", inputs={...}, outputs=[...] ) callback_handler_code = ' import boto3 import json def handler(event, context): sagemaker_client=boto3.client("sagemaker") for record in event["Records"]: payload=json.loads(record["body"]) token=payload["token"] # Custom processing # Call SageMaker AI to complete the step sagemaker_client.send_pipeline_execution_step_success( CallbackToken=token, OutputParameters={...} ) ' ``` **注意** 不得嵌套 `CallbackStep` 的输出参数。例如，如果您使用嵌套字典作为输出参数，则该字典将被视为单个字符串（例如 `{"output1": "{\"nested_output1\":\"my-output\"}"}`)。 如果您提供嵌套值，则当您尝试引用特定的输出参数时， SageMaker AI 会抛出不可重试的客户端错误。 **停止行为** 当 `Callback` 步骤运行时，管道进程不会停止。 当您使用正在运行的`Callback`步骤调用[StopPipelineExecution](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_StopPipelineExecution.html)管道进程时，Pipelines 会向 SQS 队列发送一条 Amazon SQS 消息。SQS 消息正文包含一个**状态**字段，该字段设置为 `Stopping`。以下是 SQS 消息正文示例。 ``` { "token": "26vcYbeWsZ", "pipelineExecutionArn": "arn:aws:sagemaker:us-east-2:012345678901:pipeline/callback-pipeline/execution/7pinimwddh3a", "arguments": { "number": 5, "stringArg": "some-arg", "inputData": "s3://sagemaker-us-west-2-012345678901/abalone/abalone-dataset.csv" }, "status": "Stopping" } ``` 您应该在 Amazon SQS 消息用户中添加逻辑，以便在收到消息后采取任何必要的措施（例如，资源清理）。然后添加对 `SendPipelineExecutionStepSuccess` 或 `SendPipelineExecutionStepFailure` 的调用。 只有当 Pipelines 收到这些调用时，它才会停止管道进程。 ## Lambda 步骤 Lambda 您可以使用 Lambda 步骤来运行函数。 Amazon Lambda 您可以运行现有的 Lambda 函数，或者 SageMaker AI 可以创建并运行新的 Lambda 函数。如果您选择使用现有 Lambda 函数，则该函数必须与 SageMaker AI 管 Amazon Web Services 区域 道相同。[有关展示如何在 SageMaker AI 管道中使用 Lambda 步骤的笔记本，请参阅 sagemaker-pipelines-lambda-step .ipynb。](https://github.com/aws/amazon-sagemaker-examples/blob/main/sagemaker-pipelines/tabular/lambda-step/sagemaker-pipelines-lambda-step.ipynb) **重要** 亚马逊 Pyth SageMaker on SDK v2.51.0 和亚马 SageMaker 逊 Studio Classic v3.9.1 中引入了 Lambda 步骤。您必须在使用 Lambda 步骤之前更新 Studio Classic，否则不会显示管道 DAG。要更新 Studio Classic，请参阅 [关闭并更新 Amazon SageMaker Studio 经典版](studio-tasks-update-studio.md)。 SageMaker AI 提供了 [sagemaker.lambda\$1Helper.Lambda 类来创建、更新、调用和删除 Lambda 函](https://sagemaker.readthedocs.io/en/stable/api/utility/lambda_helper.html)数。 `Lambda`具有以下签名。 ``` Lambda( function_arn, # Only required argument to invoke an existing Lambda function # The following arguments are required to create a Lambda function: function_name, execution_role_arn, zipped_code_dir, # Specify either zipped_code_dir and s3_bucket, OR script s3_bucket, # S3 bucket where zipped_code_dir is uploaded script, # Path of Lambda function script handler, # Lambda handler specified as "lambda_script.lambda_handler" timeout, # Maximum time the Lambda function can run before the lambda step fails ... ) ``` sagemaker. [workflow.lambda\$1step。 LambdaStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.lambda_step.LambdaStep)类的`lambda_func`参数类型为`Lambda`。要调用现有 Lambda 函数，唯一的要求是将函数的 Amazon 资源名称 (ARN) 提供给 `function_arn`。如果不提供 `function_arn` 的值，则必须指定 `handler` 和以下内容之一： + `zipped_code_dir` - 压缩后的 Lambda 函数的路径 `s3_bucket` - 要上传 `zipped_code_dir` 的 Amazon S3 存储桶 + `script` - Lambda 函数脚本文件的路径 以下示例说明如何创建调用现有 Lambda 函数的 `Lambda` 步骤定义。 ``` from sagemaker.workflow.lambda_step import LambdaStep from sagemaker.lambda_helper import Lambda step_lambda = LambdaStep( name="ProcessingLambda", lambda_func=Lambda( function_arn="arn:aws:lambda:us-west-2:012345678910:function:split-dataset-lambda" ), inputs={ s3_bucket = s3_bucket, data_file = data_file }, outputs=[ "train_file", "test_file" ] ) ``` 以下示例说明如何创建 `Lambda` 步骤定义，该定义使用 Lambda 函数脚本创建和调用 Lambda 函数。 ``` from sagemaker.workflow.lambda_step import LambdaStep from sagemaker.lambda_helper import Lambda step_lambda = LambdaStep( name="ProcessingLambda", lambda_func=Lambda( function_name="split-dataset-lambda", execution_role_arn=execution_role_arn, script="lambda_script.py", handler="lambda_script.lambda_handler", ... ), inputs={ s3_bucket = s3_bucket, data_file = data_file }, outputs=[ "train_file", "test_file" ] ) ``` **输入和输出** 如果您的 `Lambda` 函数有输入或输出，则还必须在 `Lambda` 步骤中定义这些输入或输出。 **注意** 不得嵌套输入和输出参数。例如，如果您使用嵌套字典作为输出参数，则该字典将被视为单个字符串（例如 `{"output1": "{\"nested_output1\":\"my-output\"}"}`）。如果您提供嵌套值并尝试稍后再引用该值，则会引发不可重试的客户端错误。 定义 `Lambda` 步骤时，`inputs` 必须是键值对的字典。`inputs` 字典的每个值都必须是基元类型（字符串、整数或浮点数）。不支持嵌套对象。如果 `inputs` 值未定义，则默认为 `None`。 `outputs` 值必须是键列表。这些键是指 `Lambda` 函数输出中定义的字典。比如 `inputs`，这些键必须是基元类型，并且不支持嵌套对象。 **超时和停止行为** `Lambda` 类有一个 `timeout` 参数，用于指定 Lambda 函数可以运行的最长时间。默认值为 120 秒，最大值为 10 分钟。如果在达到超时条件时 Lambda 函数正在运行，则 Lambda 步骤将失败；但 Lambda 函数会继续运行。 当 Lambda 步骤正在运行时，无法停止管道进程，因为无法停止 Lambda 步骤调用的 Lambda 函数。如果您在 Lambda 函数运行时停止进程，管道会等待函数完成或超时。这取决于哪个先发生。进程随即停止。如果 Lambda 函数完成，则管道进程状态为 `Stopped`。如果达到超时，则管道进程状态为 `Failed`。 ## ClarifyCheck 步 `ClarifyCheck` 您可以使用 `ClarifyCheck` 步骤对照先前的基准进行基准偏差检查，以进行偏差分析和实现模型可解释性。然后，您可以使用 `model.register()` 方法生成和[注册基准](https://docs.amazonaws.cn/sagemaker/latest/dg/pipelines-quality-clarify-baseline-lifecycle.html#pipelines-quality-clarify-baseline-calculations)，并使用 `[step\$1args](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#model-step)` 将该方法的输出传递给 [模型步骤](#step-type-model)。Amazon SageMaker 模型监控器可以将这些偏差检查基准用于您的模型终端节点。因此，您无需单独提出[基线](https://docs.amazonaws.cn/sagemaker/latest/dg/model-monitor-create-baseline.html)建议。 `ClarifyCheck` 步骤还可以从模型注册表中提取偏差检查基准。该`ClarifyCheck`步骤使用 Clari SageMaker fy 预建容器。该容器提供一系列模型监测功能，包括约束建议和根据给定基线进行约束验证。有关更多信息，请参阅 [预建的 SageMaker 澄清容器](clarify-processing-job-configure-container.md)。 ### 配置 ClarifyCheck 步骤 配置 `ClarifyCheck` 您可以将 `ClarifyCheck` 步骤配置为每次在管道中使用时仅执行以下一种类型的检查。 + 数据偏差检查 + 模型偏差检查 + 模型可解释性检查 为此，请将 `clarify_check_config` 参数设置为以下检查类型值之一： + `DataBiasCheckConfig` + `ModelBiasCheckConfig` + `ModelExplainabilityCheckConfig` 该`ClarifyCheck`步骤启动一个处理作业，该作业运行 SageMaker AI Clarify 预建容器，并且需要[为支票和处理作业进行专用配置](https://docs.amazonaws.cn/sagemaker/latest/dg/clarify-configure-processing-jobs.html)。 `ClarifyCheckConfig`和`CheckJobConfig`是这些配置的辅助函数。这些辅助函数与 Clarify 处理 SageMaker 作业的计算方式一致，用于检查模型偏差、数据偏差或模型可解释性。有关更多信息，请参阅 [运行 Cl SageMaker arify 处理作业以实现偏见分析和可解释性](clarify-processing-job-run.md)。 ### 控制偏差检查的步骤行为 配置 `ClarifyCheck` `ClarifyCheck` 步骤需要以下两个布尔标志来控制其行为： + `skip_check`：此参数表示是否跳过针对先前基准的偏差检查。如果将其设置为 `False`，则配置的检查类型的先前基准必须可用。 + `register_new_baseline`：此参数表示是否可以通过步骤属性 `BaselineUsedForDriftCheckConstraints` 访问新计算的基准。如果将其设置为 `False`，则配置的检查类型的先前基准也必须可用。这可以通过 `BaselineUsedForDriftCheckConstraints` 属性访问。 有关更多信息，请参阅 [基准计算、偏差检测和生命周期以及 Amazon Pi ClarifyCheck pelin SageMaker es 中的 QualityCheck 步骤](pipelines-quality-clarify-baseline-lifecycle.md)。 ### 使用基准 基准 您可以选择指定 `model_package_group_name` 来定位现有基线。然后，`ClarifyCheck` 步骤在模型软件包组中最新批准的模型软件包上提取 `DriftCheckBaselines`。 或者，您可以通过 `supplied_baseline_constraints` 参数提供先前的基准。如果同时指定 `model_package_group_name` 和 `supplied_baseline_constraints`，则 `ClarifyCheck` 步骤将使用 `supplied_baseline_constraints` 参数指定的基准。 有关使用`ClarifyCheck`步骤要求的更多信息，请参阅 [sagemaker.workflow.steps。 ClarifyCheckStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.clarify_check_step.ClarifyCheckStep)在适用于 *Python 的亚马逊 SageMaker AI SageMaker 人工智能开发工具包*中。有关展示如何使用 Pipelines 中`ClarifyCheck`步骤的 Amazon SageMaker Studio Classic 笔记本，请参阅 [sagemaker-pipeline-model-monitor-clarify-steps.ipynb](https://github.com/aws/amazon-sagemaker-examples/blob/main/sagemaker-pipelines/tabular/model-monitor-clarify-pipelines/sagemaker-pipeline-model-monitor-clarify-steps.ipynb)。 **Example 创建 `ClarifyCheck` 步骤以进行数据偏差检查** ``` from sagemaker.workflow.check_job_config import CheckJobConfig from sagemaker.workflow.clarify_check_step import DataBiasCheckConfig, ClarifyCheckStep from sagemaker.workflow.execution_variables import ExecutionVariables check_job_config = CheckJobConfig( role=role, instance_count=1, instance_type="ml.c5.xlarge", volume_size_in_gb=120, sagemaker_session=sagemaker_session, ) data_bias_data_config = DataConfig( s3_data_input_path=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri, s3_output_path=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'databiascheckstep']), label=0, dataset_type="text/csv", s3_analysis_config_output_path=data_bias_analysis_cfg_output_path, ) data_bias_config = BiasConfig( label_values_or_threshold=[15.0], facet_name=[8], facet_values_or_threshold=[[0.5]] ) data_bias_check_config = DataBiasCheckConfig( data_config=data_bias_data_config, data_bias_config=data_bias_config, )h data_bias_check_step = ClarifyCheckStep( name="DataBiasCheckStep", clarify_check_config=data_bias_check_config, check_job_config=check_job_config, skip_check=False, register_new_baseline=False supplied_baseline_constraints="s3://sagemaker-us-west-2-111122223333/baseline/analysis.json", model_package_group_name="MyModelPackageGroup" ) ``` ## QualityCheck 步 `QualityCheck` 使用 `QualityCheck` 步骤对管道中的数据质量或模型质量进行[基线建议](https://docs.amazonaws.cn/sagemaker/latest/dg/model-monitor-create-baseline.html)和漂移检查。然后，您可以使用 `model.register()` 方法生成和[注册基准](https://docs.amazonaws.cn/sagemaker/latest/dg/pipelines-quality-clarify-baseline-lifecycle.html#pipelines-quality-clarify-baseline-calculations)，并使用 `[step\$1args](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#model-step)` 将该方法的输出传递给 [模型步骤](#step-type-model)。 Model Monitor 可以将这些偏差检查基准用于模型端点，这样您就无需单独提出基准建议。`QualityCheck` 步骤还可以从模型注册表中提取偏差检查基准。该`QualityCheck`步骤利用了 Amazon A SageMaker I 模型监控器预先构建的容器。该容器具有一系列模型监测功能，包括约束建议、统计数据生成和对照基线进行约束验证。有关更多信息，请参阅 [Amazon SageMaker 模型监控器预建容器](model-monitor-pre-built-container.md)。 ### 配置 QualityCheck 步骤 配置 `QualityCheck` 您可以对 `QualityCheck` 步骤进行配置，使其每次在管道中使用时只运行以下一种检查类型。 + 数据质量检查 + 模型质量检查 您可以通过将 `quality_check_config` 参数设置为以下检查类型值之一来实现此目的： + `DataQualityCheckConfig` + `ModelQualityCheckConfig` `QualityCheck` 步骤会启动一个处理作业，该作业运行 Model Monitor 预构建容器，并且需要专门的用于检查和处理作业的配置。`QualityCheckConfig` 和 `CheckJobConfig` 是这些配置的辅助功能。这些辅助功能与模型监控器为模型质量或数据质量监控创建基线的方式一致。有关 Model Monitor 基准建议的更多信息，请参阅[创建基准](model-monitor-create-baseline.md)和[创建模型质量基线](model-monitor-model-quality-baseline.md)。 ### 控制偏差检查的步骤行为 配置 `QualityCheck` `QualityCheck` 步骤需要以下两个布尔标志来控制其行为： + `skip_check`：此参数表示是否跳过针对先前基准的偏差检查。如果将其设置为 `False`，则配置的检查类型的先前基准必须可用。 + `register_new_baseline`：此参数表示是否可以通过步骤属性 `BaselineUsedForDriftCheckConstraints` 和 `BaselineUsedForDriftCheckStatistics` 访问新计算的基准。如果将其设置为 `False`，则配置的检查类型的先前基准也必须可用。它们可以通过 `BaselineUsedForDriftCheckConstraints` 和 `BaselineUsedForDriftCheckStatistics` 属性进行访问。 有关更多信息，请参阅[基准计算、偏差检测和生命周期以及 Amazon Pi ClarifyCheck pelin SageMaker es 中的 QualityCheck 步骤](pipelines-quality-clarify-baseline-lifecycle.md)。 ### 使用基准 基准 您可以通过 `supplied_baseline_statistics` 和 `supplied_baseline_constraints` 参数直接指定上一条基线。您还可以指定 `model_package_group_name` 和 `QualityCheck` 步骤，在模型软件包组中最新批准的模型软件包上拉动 `DriftCheckBaselines`。 指定以下内容时，`QualityCheck` 步骤将使用 `supplied_baseline_constraints` 和 `supplied_baseline_statistics` 在 `QualityCheck` 步骤的校验类型上指定的基线。 + `model_package_group_name` + `supplied_baseline_constraints` + `supplied_baseline_statistics` 有关使用`QualityCheck`步骤要求的更多信息，请参阅 [sagemaker.workflow.steps。 QualityCheckStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.quality_check_step.QualityCheckStep)在适用于 *Python 的亚马逊 SageMaker AI SageMaker 人工智能开发工具包*中。有关展示如何使用 Pipelines 中`QualityCheck`步骤的 Amazon SageMaker Studio Classic 笔记本，请参阅 [sagemaker-pipeline-model-monitor-clarify-steps.ipynb](https://github.com/aws/amazon-sagemaker-examples/blob/main/sagemaker-pipelines/tabular/model-monitor-clarify-pipelines/sagemaker-pipeline-model-monitor-clarify-steps.ipynb)。 **Example 创建 `QualityCheck` 步骤以进行数据质量检查** ``` from sagemaker.workflow.check_job_config import CheckJobConfig from sagemaker.workflow.quality_check_step import DataQualityCheckConfig, QualityCheckStep from sagemaker.workflow.execution_variables import ExecutionVariables check_job_config = CheckJobConfig( role=role, instance_count=1, instance_type="ml.c5.xlarge", volume_size_in_gb=120, sagemaker_session=sagemaker_session, ) data_quality_check_config = DataQualityCheckConfig( baseline_dataset=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri, dataset_format=DatasetFormat.csv(header=False, output_columns_position="START"), output_s3_uri=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'dataqualitycheckstep']) ) data_quality_check_step = QualityCheckStep( name="DataQualityCheckStep", skip_check=False, register_new_baseline=False, quality_check_config=data_quality_check_config, check_job_config=check_job_config, supplied_baseline_statistics="s3://sagemaker-us-west-2-555555555555/baseline/statistics.json", supplied_baseline_constraints="s3://sagemaker-us-west-2-555555555555/baseline/constraints.json", model_package_group_name="MyModelPackageGroup" ) ``` ## EMR 步骤 EMR 使用 Amazon P [ip SageMaker elines EMR 步骤可](https://docs.amazonaws.cn/emr/latest/ManagementGuide/emr-overview.html)以： + 在运行中的 Amazon EMR 集群上处理 [Amazon EMR 步骤](https://docs.amazonaws.cn/emr/latest/ManagementGuide/emr-work-with-steps.html)。 + 让管道为您创建和管理 Amazon EMR 集群。 有关 Amazon EMR 的更多信息，请参阅 [Amazon EMR 入门](https://docs.amazonaws.cn/emr/latest/ManagementGuide/emr-gs.html)。 EMR 步骤要求 `EMRStepConfig` 包括 Amazon EMR 集群使用的 JAR 文件的位置以及要传递的任何参数。如果您要在运行中的 EMR 集群上运行该步骤，还需提供 Amazon EMR 集群 ID。您还可以通过集群配置，在它为您创建、管理和终止的集群上运行 EMR 步骤。以下几节包括演示这两种方法的示例笔记本的示例和链接。 **注意** EMR 步骤要求传递给管道的角色具有其他权限。将 [Amazon 托管式策略附加到管道角色：`AmazonSageMakerPipelinesIntegrations`](https://docs.amazonaws.cn/sagemaker/latest/dg/security-iam-awsmanpol-pipelines.html#security-iam-awsmanpol-AmazonSageMakerPipelinesIntegrations) 到管道角色，或确保角色包含该策略中的权限。 如果您在运行中的集群上处理 EMR 步骤，则只能使用处于以下状态之一的集群： `STARTING` `BOOTSTRAPPING` `RUNNING` `WAITING` 如果您在正在运行的集群上处理 EMR 步骤，则在 EMR 集群上最多可以有 256 个处于 `PENDING` 状态的 EMR 步骤。提交的 EMR 步骤超过此限制会导致管道执行失败。您可以考虑使用[管道步骤的重试策略](pipelines-retry-policy.md)。 您可以指定集群 ID 或集群配置，但不能同时指定两者。 EMR 步骤依靠 Amazon EventBridge 来监控 EMR 步骤或集群状态的变化。如果您在正在运行的集群上处理 Amazon EMR 作业，则 EMR 步骤将使用 `SageMakerPipelineExecutionEMRStepStatusUpdateRule` 规则来监控 EMR 步骤的状态。如果您在 EMR 步骤创建的集群上处理作业，则该步骤会使用 `SageMakerPipelineExecutionEMRClusterStatusRule` 规则监控集群状态的变化。如果您在 Amazon 账户中看到其中任何一条 EventBridge 规则，请不要将其删除，否则您的 EMR 步骤可能无法完成。 **在您的管道中添加 Amazon EMR 步骤** 要向您的管道添加 EMR 步骤，请执行以下操作： + 按照[启动 Amazon Studio 中的说明打开 SageMaker Studio](https://docs.amazonaws.cn/sagemaker/latest/dg/studio-updated-launch.html) 控制台。 + 在左侧导航窗格中，选择 **Pipelines**。 + 选择**创建**。 + 选择**空白**。 + 在左侧边栏中选择**处理数据**，然后将其拖到画布上。 + 在画布中，选择添加的**处理数据**步骤。 + 在右侧边栏的 “模式” 下，选择 “**EMR（托管）**”。 + 在右侧边栏中，填写 “**设置” 和 “详细信息**” 选项卡中的表单。有关这些选项卡中字段的信息，请参阅 [sagemaker.workflow.fail\$1step。 EMRstep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.emr_step.EMRStep)。 **在正在运行的 Amazon EMR 集群上启动新作业** 要在运行中的 Amazon EMR 集群上启动新作业，请将集群 ID 作为字符串传递给 `EMRStep` 的 `cluster_id` 参数。以下示例演示了此过程。 ``` from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig emr_config = EMRStepConfig( jar="jar-location", # required, path to jar file used args=["--verbose", "--force"], # optional list of arguments to pass to the jar main_class="com.my.Main1", # optional main class, this can be omitted if jar above has a manifest properties=[ # optional list of Java properties that are set when the step runs { "key": "mapred.tasktracker.map.tasks.maximum", "value": "2" }, { "key": "mapreduce.map.sort.spill.percent", "value": "0.90" }, { "key": "mapreduce.tasktracker.reduce.tasks.maximum", "value": "5" } ] ) step_emr = EMRStep ( name="EMRSampleStep", # required cluster_id="j-1ABCDEFG2HIJK", # include cluster_id to use a running cluster step_config=emr_config, # required display_name="My EMR Step", description="Pipeline step to execute EMR job" ) ``` 有关指导您完成完整示例的示例笔记本，请参阅[运行 EMR 集群的 Pipelines EMR 步骤](https://github.com/aws/amazon-sagemaker-examples/blob/main/sagemaker-pipelines/tabular/emr-step/sagemaker-pipelines-emr-step-with-running-emr-cluster.ipynb)。 **在新的 Amazon EMR 集群上启动新作业** 要在 `EMRStep` 为您创建的新集群上启动新作业，请将集群配置作为字典提供。字典必须与[RunJobFlow](https://docs.amazonaws.cn/emr/latest/APIReference/API_RunJobFlow.html)请求具有相同的结构。但是，请勿在集群配置中包含以下字段： + [`Name`] + [`Steps`] + [`AutoTerminationPolicy`] + [`Instances`][`KeepJobFlowAliveWhenNoSteps`] + [`Instances`][`TerminationProtected`] 所有其他 `RunJobFlow` 参数均可在您的集群配置中使用。有关请求语法的详细信息，请参阅[RunJobFlow](https://docs.amazonaws.cn/emr/latest/APIReference/API_RunJobFlow.html)。 下面的示例将集群配置传递给 EMR 步骤定义。这将提示在新的 EMR 集群上启动新作业的步骤。此示例中的 EMR 集群配置包括主节点和核心 EMR 集群节点的规范。有关 Amazon EMR 节点类型的更多信息，请参阅[了解节点类型：主节点、核心节点和任务节点](https://docs.amazonaws.cn/emr/latest/ManagementGuide/emr-master-core-task-nodes.html)。 ``` from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig emr_step_config = EMRStepConfig( jar="jar-location", # required, path to jar file used args=["--verbose", "--force"], # optional list of arguments to pass to the jar main_class="com.my.Main1", # optional main class, this can be omitted if jar above has a manifest properties=[ # optional list of Java properties that are set when the step runs { "key": "mapred.tasktracker.map.tasks.maximum", "value": "2" }, { "key": "mapreduce.map.sort.spill.percent", "value": "0.90" }, { "key": "mapreduce.tasktracker.reduce.tasks.maximum", "value": "5" } ] ) # include your cluster configuration as a dictionary emr_cluster_config = { "Applications": [ { "Name": "Spark", } ], "Instances":{ "InstanceGroups":[ { "InstanceRole": "MASTER", "InstanceCount": 1, "InstanceType": "m5.2xlarge" }, { "InstanceRole": "CORE", "InstanceCount": 2, "InstanceType": "m5.2xlarge" } ] }, "BootstrapActions":[], "ReleaseLabel": "emr-6.6.0", "JobFlowRole": "job-flow-role", "ServiceRole": "service-role" } emr_step = EMRStep( name="emr-step", cluster_id=None, display_name="emr_step", description="MyEMRStepDescription", step_config=emr_step_config, cluster_config=emr_cluster_config ) ``` 有关指导您完成完整示例的示例笔记本，请参阅[使用集群生命周期管理的管道 EMR 步骤](https://github.com/aws/amazon-sagemaker-examples/blob/main/sagemaker-pipelines/tabular/emr-step/sagemaker-pipelines-emr-step-with-cluster-lifecycle-management.ipynb)。 ## EMR 无服务器步骤 EMR 无服务器步骤 要向您的管道中添加 EMR 无服务器步骤，请执行以下操作： + 按照[启动 Amazon Studio 中的说明打开 SageMaker Studio](https://docs.amazonaws.cn/sagemaker/latest/dg/studio-updated-launch.html) 控制台。 + 在左侧导航窗格中，选择 **Pipelines**。 + 选择**创建**。 + 选择**空白**。 + 在左侧边栏中选择**处理数据**，然后将其拖到画布上。 + 在画布中，选择添加的**处理数据**步骤。 + 在右侧边栏的模式下，选择 **EMR（无服务器）**。 + 在右侧边栏中，填写 “**设置” 和 “详细信息**” 选项卡中的表单。 ## 笔记本作业步骤 笔记本作业 使用`NotebookJobStep`以非交互方式运行您的 SageMaker Notebook Job 作为工作流步骤。如果您在 Pipelines drag-and-drop 用户界面中构建管道，请使用[执行代码步骤](#step-type-executecode)来运行您的笔记本。有关 SageMaker 笔记本作业的更多信息，请参阅[SageMaker 笔记本职位](notebook-auto-run.md)。 `NotebookJobStep` 至少需要输入笔记本、映像 URI 和内核名称。有关 Notebook Job 步骤要求以及可以设置为自定义步骤的其他参数的更多信息，请参阅 [sagemaker.workflow.steps。 NotebookJobStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.notebook_job_step.NotebookJobStep)。 下面的示例使用最小参数定义了 `NotebookJobStep`。 ``` from sagemaker.workflow.notebook_job_step import NotebookJobStep notebook_job_step = NotebookJobStep( input_notebook=input_notebook, image_uri=image_uri, kernel_name=kernel_name ) ``` 您的工作`NotebookJobStep`流步骤被视为 SageMaker 笔记本作业。因此，在 Studio Classic UI 笔记本作业控制面板中，通过使用 `tags` 参数包含特定标签来跟踪执行状态。有关包含标签的更多详情，请参阅 [在 Studio UI 面板上查看笔记本作业](create-notebook-auto-run-sdk.md#create-notebook-auto-run-dash)。 此外，如果您使用 SageMaker Python SDK 安排笔记本作业，则只能指定某些图像来运行笔记本作业。有关更多信息，请参阅 [SageMaker AI Python SDK 笔记本作业的图像限制](notebook-auto-run-constraints.md#notebook-auto-run-constraints-image-sdk)。 ## Fail 步骤 Fail 当未达到所需条件或状态时，使用 “失败” 步骤停止 Amazon Pipelin SageMaker es 的执行。在 Fail 步骤中还可以输入自定义错误信息，说明管道执行失败的原因。 **注意** 当一个 Fail 步骤和其他管道步骤同时执行时，管道在所有并发步骤完成之前不会终止。 ### 使用 Fail 步骤的限制 Fail 步骤限制 + 您不能将 Fail 步骤添加到其他步骤的 `DependsOn` 列表中。有关更多信息，请参阅 [步骤之间的自定义依赖关系](build-and-manage-steps.md#build-and-manage-custom-dependency)。 + 其他步骤不能引用 Fail 步骤。它*始终* 是管道执行的最后一步。 + 您不能重试以 Fail 步骤结束的管道执行。 您可以创建静态文本字符串形式的 Fail 步骤错误信息。另外，如果您使用 SDK，也可以使用 [Pipeline Parameters](https://docs.amazonaws.cn/sagemaker/latest/dg/build-and-manage-parameters.html)、[Join](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html?highlight=Join#sagemaker.workflow.functions.Join) 操作或其他[步骤属性](https://docs.amazonaws.cn/sagemaker/latest/dg/build-and-manage-steps.html#build-and-manage-properties)来创建信息量更大的错误信息。 ------ #### [ Pipeline Designer ] 要在管道中添加 Fail 步骤，请执行以下操作： 1. 按照 [启动亚马逊 SageMaker Studio](studio-updated-launch.md) 中的说明打开 Studio 管理控制台。 1. 在左侧导航窗格中，选择 **Pipelines**。 1. 选择**创建**。 1. 选择**空白**。 1. 在左侧边栏中选择 **Fail**，然后将其拖到画布上。 1. 在画布中，选择添加的 **Fail** 步骤。 1. 在右侧边栏中，填写**设置**和**详情**标签中的表格。有关这些选项卡中字段的信息，请参阅 [sagemaker.workflow.fail\$1step。 FailStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.fail_step.FailStep)。 1. 如果画布上有任何步骤紧接在您添加的 **Fail** 步骤之前，请单击并拖动光标从该步骤到 **Fail** 步骤，以创建边缘。 1. 如果画布上有任何步骤紧接着您添加的 **Fail** 步骤，请单击并拖动光标从 **Fail** 步骤到该步骤，以创建边缘。 ------ #### [ SageMaker Python SDK ] **Example** 以下示例代码片段使用了一个 `FailStep`（带有 `ErrorMessage`，并配置了管道参数）和一项 `Join` 操作。 ``` from sagemaker.workflow.fail_step import FailStep from sagemaker.workflow.functions import Join from sagemaker.workflow.parameters import ParameterInteger mse_threshold_param = ParameterInteger(name="MseThreshold", default_value=5) step_fail = FailStep( name="AbaloneMSEFail", error_message=Join( on=" ", values=["Execution failed due to MSE >", mse_threshold_param] ), ) ``` ------ # 添加集成 MLflow 集成允许您 MLflow 与管道一起使用来选择跟踪服务器或无服务器应用程序、选择实验和记录指标。 ## 重要概念 **默认应用程序创建**-当您进入管道可视化编辑器时，将创建默认 MLflow 应用程序。 **集成面板**-包含一个新的集成面板 MLflow，您可以选择和配置该面板。 **更新应用程序和实验**-在管道执行期间覆盖所选应用程序和实验的选项。 ## 工作原理 + 前往**管道可视化编辑器** + 在工具栏上选择 “**集成**” + 选择 **MLflow** + 配置 MLflow 应用程序并进行实验 ## 屏幕截图示例 集成侧面板 ![\[待办事项描述。\]](http://docs.amazonaws.cn/sagemaker/latest/dg/images/screenshot-pipeline-1.png) MLflow 配置 ![\[待办事项描述。\]](http://docs.amazonaws.cn/sagemaker/latest/dg/images/screenshot-pipeline-2.png) 如何在管道执行期间重写实验 ![\[待办事项描述。\]](http://docs.amazonaws.cn/sagemaker/latest/dg/images/screenshot-pipeline-3.png) ## 步骤属性 使用 `properties` 属性在管道中的步骤之间添加数据依赖关系。管道使用这些数据依赖关系来根据管道定义构建 DAG。这些属性可以作为占位符值引用，并在运行时解析。 Pi `properties` pelines 步骤的属性与`Describe`调用相应 SageMaker AI 作业类型返回的对象相匹配。对于每种作业类型，`Describe` 调用都会返回以下响应对象： + `ProcessingStep` – [DescribeProcessingJob](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_DescribeProcessingJob.html) + `TrainingStep` – [DescribeTrainingJob](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_DescribeTrainingJob.html) + `TransformStep` – [DescribeTransformJob](https://docs.amazonaws.cn/sagemaker/latest/APIReference/API_DescribeTransformJob.html) 要在创建数据依赖关系期间查看每种步骤类型哪些属性是可引用的，请参阅 [Amaz SageMaker on Python SDK](https://sagemaker.readthedocs.io/en/stable) 中的*[数据依赖关系-属性参考](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#data-dependency-property-reference)*。 ## 步骤并行 当一个步骤不依赖于任何其他步骤时，它会在管道执行后立即运行。但是，并行执行过多的管道步骤可能会很快耗尽可用资源。使用 `ParallelismConfiguration` 控制管道执行的并发步骤数。 以下示例使用 `ParallelismConfiguration` 将并发步骤数限制设置为 5。 ``` pipeline.create( parallelism_config=ParallelismConfiguration(5), ) ``` ## 步骤之间的数据依赖性 您可以通过指定步骤之间的数据关系来定义 DAG 的结构。要在步骤之间创建数据依赖关系，请将一个步骤的属性作为输入传递给管道中的另一个步骤。接收输入的步骤要等到提供输入的步骤完成运行后才会开始。 数据依赖关系使用以下格式的 JsonPath 符号。这种格式会遍历 JSON 属性文件。这意味着您可以根据需要追加任意数量的**实例，以达到文件中所需的嵌套属性。有关 JsonPath 符号的更多信息，请参阅 [JsonPath repo](https://github.com/json-path/JsonPath)。 ``` .properties.. ``` 以下内容说明了如何使用处理步骤的 `ProcessingOutputConfig` 属性指定 Amazon S3 存储桶。 ``` step_process.properties.ProcessingOutputConfig.Outputs["train_data"].S3Output.S3Uri ``` 要创建数据依赖关系，请按如下方式将存储桶传递给训练步骤。 ``` from sagemaker.workflow.pipeline_context import PipelineSession sklearn_train = SKLearn(..., sagemaker_session=PipelineSession()) step_train = TrainingStep( name="CensusTrain", step_args=sklearn_train.fit(inputs=TrainingInput( s3_data=step_process.properties.ProcessingOutputConfig.Outputs[ "train_data"].S3Output.S3Uri )) ) ``` 要在创建数据依赖关系期间查看每种步骤类型哪些属性是可引用的，请参阅 [Amaz SageMaker on Python SDK](https://sagemaker.readthedocs.io/en/stable) 中的*[数据依赖关系-属性参考](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#data-dependency-property-reference)*。 ## 步骤之间的自定义依赖关系 指定数据依赖关系时，Pipelines 会在各步骤之间提供数据连接。另外，一个步骤可以访问前一个步骤的数据，而无需直接使用管道。在这种情况下，您可以创建一个自定义依赖关系，告诉 Pipelines 在另一个步骤运行完之前不要启动该步骤。您可以通过指定步骤的 `DependsOn` 属性来创建自定义依赖关系。 例如，以下内容定义了步骤 `C`，该步骤在步骤 `A` 和步骤 `B` 都完成运行后才开始。 ``` { 'Steps': [ {'Name':'A', ...}, {'Name':'B', ...}, {'Name':'C', 'DependsOn': ['A', 'B']} ] } ``` 如果依赖关系会产生循环依赖关系，Pipelines 会抛出验证异常。 以下示例创建了一个在处理步骤完成运行后开始的训练步骤。 ``` processing_step = ProcessingStep(...) training_step = TrainingStep(...) training_step.add_depends_on([processing_step]) ``` 以下示例创建了一个训练步骤，该步骤要等到两个不同的处理步骤完成运行后才会开始。 ``` processing_step_1 = ProcessingStep(...) processing_step_2 = ProcessingStep(...) training_step = TrainingStep(...) training_step.add_depends_on([processing_step_1, processing_step_2]) ``` 下面提供了另一种创建自定义依赖关系的方法。 ``` training_step.add_depends_on([processing_step_1]) training_step.add_depends_on([processing_step_2]) ``` 以下示例创建了一个训练步骤，该步骤接收来自一个处理步骤的输入，然后等待另一个处理步骤完成运行。 ``` processing_step_1 = ProcessingStep(...) processing_step_2 = ProcessingStep(...) training_step = TrainingStep( ..., inputs=TrainingInput( s3_data=processing_step_1.properties.ProcessingOutputConfig.Outputs[ "train_data" ].S3Output.S3Uri ) training_step.add_depends_on([processing_step_2]) ``` 以下示例说明如何检索步骤的自定义依赖关系的字符串列表。 ``` custom_dependencies = training_step.depends_on ``` ## 自定义映像一步到位 在管道中创建步骤时，您可以使用任何可用的 SageMaker AI [深度学习容器镜像](https://github.com/aws/deep-learning-containers/blob/master/available_images.md)。 您还可以在管道步骤中使用自己的容器。由于无法在 Studio Classic 中创建映像，因此在使用 Pipelines 之前，必须使用其他方法创建映像。 要在为管道创建步骤时使用自己的容器，请在估算器定义中包含映像 URI。有关将自己的容器与 SageMaker AI 配合使用的更多信息，请参阅将 [Docker 容器与 SageMaker AI 配合使用](https://docs.amazonaws.cn/sagemaker/latest/dg/docker-containers.html)。 # Lift-and-shift 使用 @step 装饰器的 Python 代码 `@step` 装饰器是将本地机器学习 (ML) 代码转换为一个或多个管道步骤的功能。您可以像编写任何 ML 项目一样编写 ML 函数。在本地测试或使用装饰器作为训练作业进行测试后，您可以通过添加`@remote`装饰器将该函数转换为 A SageMaker I 管道步骤。`@step`然后，您可以将 `@step` 装饰函数调用的输出作为一个步骤传递给 Pipelines，以创建并运行管道。您还可以使用 `@step` 装饰器串联一系列函数，创建多步骤有向无环图 (DAG) 管道。 使用 `@step` 装饰器的设置与使用 `@remote` 装饰器的设置相同。关于如何[设置环境](https://docs.amazonaws.cn/sagemaker/latest/dg/train-remote-decorator.html#train-remote-decorator-env)和[使用配置文件](https://docs.amazonaws.cn/sagemaker/latest/dg/train-remote-decorator-config.html)设置默认值，请参阅远程功能文档。有关 `@step` 装饰器的更多信息，请参阅 [sagemaker.workflow.function\$1step.step](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.function_step.step)。 要查看演示如何使用 `@step` 装饰器的示例笔记本，请参阅 [@step 装饰器示例笔记本](https://github.com/aws/amazon-sagemaker-examples/tree/main/sagemaker-pipelines/step-decorator)。 下文将介绍如何使用 `@step` 装饰器注释本地 ML 代码以创建步骤、使用步骤创建和运行管道，以及如何根据使用场景定制体验。 **Topics** + [ # 使用 `@step` 装饰函数创建管道 ](pipelines-step-decorator-create-pipeline.md) + [ # 运行管道 ](pipelines-step-decorator-run-pipeline.md) + [ # 配置您的管道 ](pipelines-step-decorator-cfg-pipeline.md) + [ # 最佳实践 ](pipelines-step-decorator-best.md) + [ # 限制 ](pipelines-step-decorator-limit.md) # 使用 `@step` 装饰函数创建管道 您可以使用 `@step` 装饰器将 Python 函数转换为管道步骤，在这些函数之间创建依赖关系以创建管道图（或有向无环图 (DAG)），并将该图的叶节点作为步骤列表传递给管道，从而创建管道。下文将结合示例详细解释这一程序。 **Topics** + [ ## 将函数转换为步骤 ](#pipelines-step-decorator-run-pipeline-convert) + [ ## 在各步骤之间建立依赖关系 ](#pipelines-step-decorator-run-pipeline-link) + [ ## 使用 `ConditionStep` 和 `@step` 装饰步骤 ](#pipelines-step-decorator-condition) + [ ## 使用步骤的 `DelayedReturn` 输出定义管道 ](#pipelines-step-define-delayed) + [ ## 创建管道 ](#pipelines-step-decorator-pipeline-create) ## 将函数转换为步骤 要使用 `@step` 装饰器创建一个步骤，请使用 `@step` 对功能进行注释。下面的示例展示了一个预处理数据的 `@step` 装饰功能。 ``` from sagemaker.workflow.function_step import step @step def preprocess(raw_data): df = pandas.read_csv(raw_data) ... return procesed_dataframe step_process_result = preprocess(raw_data) ``` 当你调用`@step`装饰过的函数时， SageMaker AI 会返回一个`DelayedReturn`实例，而不是运行该函数。`DelayedReturn` 实例是该功能实际返回值的代理。`DelayedReturn` 实例可以作为参数传递给其他函数，也可以作为步骤直接传递给管道实例。有关该`DelayedReturn`课程的信息，请参阅 [sagemaker.workflow.function\$1step。 DelayedReturn](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.function_step.DelayedReturn)。 ## 在各步骤之间建立依赖关系 在两个步骤之间创建依赖关系时，就在管道图中的步骤之间创建了连接。下文将介绍在管道步骤之间创建依赖关系的多种方法。 ### 通过输入参数实现数据依赖 将一个函数的 `DelayedReturn` 输出作为另一个函数的输入，会自动在管道 DAG 中创建数据依赖关系。在下面的示例中，将 `preprocess` 函数的 `DelayedReturn` 输出传递给 `train` 函数会在 `preprocess` 和 `train` 之间产生依赖关系。 ``` from sagemaker.workflow.function_step import step @step def preprocess(raw_data): df = pandas.read_csv(raw_data) ... return procesed_dataframe @step def train(training_data): ... return trained_model step_process_result = preprocess(raw_data) step_train_result = train(step_process_result) ``` 上一个示例定义了一个训练函数，该函数用 `@step` 修饰。调用该函数时，它会接收预处理管道步骤的 `DelayedReturn` 输出作为输入。调用训练函数会返回另一个 `DelayedReturn` 实例。该实例保存了该函数中定义的所有先前步骤的信息（即本例中的 `preprocess` 步骤），这些步骤构成了管道 DAG。 在上一个示例中，`preprocess` 函数返回一个值。有关列表或元组等更复杂的返回类型，请参阅 [限制](pipelines-step-decorator-limit.md)。 ### 定义自定义依赖关系 在上一个示例中，`train` 函数接收了 `preprocess` 的 `DelayedReturn` 输出，并创建了一个依赖关系。如果想明确定义依赖关系，而不传递前一步的输出结果，请在步骤中使用 `add_depends_on` 函数。您可以使用 `get_step()` 函数从其 `DelayedReturn` 实例中获取基础步骤，然后将依赖关系作为输入调用 `add_depends_on`\$1on。要查看 `get_step()` 函数定义，请参阅 [sagemaker.workflow.step\$1outputs.get\$1step](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.step_outputs.get_step)。下面的示例演示了如何使用 `get_step()` 和 `add_depends_on()` 在 `preprocess` 和 `train` 之间创建依赖关系。 ``` from sagemaker.workflow.step_outputs import get_step @step def preprocess(raw_data): df = pandas.read_csv(raw_data) ... processed_data = .. return s3.upload(processed_data) @step def train(): training_data = s3.download(....) ... return trained_model step_process_result = preprocess(raw_data) step_train_result = train() get_step(step_train_result).add_depends_on([step_process_result]) ``` ### 将数据从 `@step` 装饰函数传递到传统管道步骤 您可以创建一个包含 `@step` 装饰步骤和传统管道步骤的管道，并在两者之间传递数据。例如，您可以使用 `ProcessingStep` 处理数据，并将处理结果传递给 `@step` 装饰的训练函数。在下面的示例中，`@step` 装饰的训练步骤引用了处理步骤的输出。 ``` # Define processing step from sagemaker.sklearn.processing import SKLearnProcessor from sagemaker.processing import ProcessingInput, ProcessingOutput from sagemaker.workflow.steps import ProcessingStep sklearn_processor = SKLearnProcessor( framework_version='1.2-1', role='arn:aws:iam::123456789012:role/SagemakerExecutionRole', instance_type='ml.m5.large', instance_count='1', ) inputs = [ ProcessingInput(source=input_data, destination="/opt/ml/processing/input"), ] outputs = [ ProcessingOutput(output_name="train", source="/opt/ml/processing/train"), ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"), ProcessingOutput(output_name="test", source="/opt/ml/processing/test") ] process_step = ProcessingStep( name="MyProcessStep", step_args=sklearn_processor.run(inputs=inputs, outputs=outputs,code='preprocessing.py'), ) ``` ``` # Define a @step-decorated train step which references the # output of a processing step @step def train(train_data_path, test_data_path): ... return trained_model step_train_result = train( process_step.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri, process_step.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri, ) ``` ## 使用 `ConditionStep` 和 `@step` 装饰步骤 管道支持一个 `ConditionStep` 类，该类会评估前几个步骤的结果，以决定在管道中采取什么行动。您也可以在 `@step` 装饰的步骤中使用 `ConditionStep`。要使用 `ConditionStep` 中任何 `@step` 装饰步骤的输出，请将该步骤的输出作为 `ConditionStep` 的参数输入。在下面的示例中，条件步骤接收 `@step` 装饰模型评测步骤的输出。 ``` # Define steps @step(name="evaluate") def evaluate_model(): # code to evaluate the model return { "rmse":rmse_value } @step(name="register") def register_model(): # code to register the model ... ``` ``` # Define ConditionStep from sagemaker.workflow.condition_step import ConditionStep from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo from sagemaker.workflow.fail_step import FailStep conditionally_register = ConditionStep( name="conditional_register", conditions=[ ConditionGreaterThanOrEqualTo( # Output of the evaluate step must be json serializable left=evaluate_model()["rmse"], # right=5, ) ], if_steps=[FailStep(name="Fail", error_message="Model performance is not good enough")], else_steps=[register_model()], ) ``` ## 使用步骤的 `DelayedReturn` 输出定义管道 无论是否使用 `@step` 装饰器，定义管道的方式都是一样的。向管道传递 `DelayedReturn` 实例时，无需传递完整的步骤列表来构建管道。SDK 会根据您定义的依赖关系自动推导出前面的步骤。您传递给管道的所有上一步骤 `Step` 对象或相关 `DelayedReturn` 对象都包含在管道图中。在下面的示例中，管道为 `train` 函数接收 `DelayedReturn` 对象。 SageMaker AI 将该`preprocess`步骤作为上一步添加到管道图中。`train` ``` from sagemaker.workflow.pipeline import Pipeline pipeline = Pipeline( name="", steps=[step_train_result], sagemaker_session=, ) ``` 如果步骤之间没有数据或自定义依赖关系，并且并行运行多个步骤，管道图就会有多个叶节点。如下例所示，将所有这些叶节点以列表形式传递给管道定义中的 `steps` 参数： ``` @step def process1(): ... return data @step def process2(): ... return data step_process1_result = process1() step_process2_result = process2() pipeline = Pipeline( name="", steps=[step_process1_result, step_process2_result], sagemaker_session=sagemaker-session, ) ``` 管道运行时，两个步骤并行。 您只需将图的叶节点传递给管道，因为叶节点包含通过数据或自定义依赖关系定义的所有先前步骤的信息。在编译管道时， SageMaker AI 还会推断出构成管道图的所有后续步骤，并将每个步骤作为单独的步骤添加到管道中。 ## 创建管道 通过调用 `pipeline.create()` 创建管道，如以下代码所示。有关 `create()` 的详细信息，请参阅 [sagemaker.workflow.pipeline.Pipeline.create](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.pipeline.Pipeline.create)。 ``` role = "pipeline-role" pipeline.create(role) ``` 当你调用时`pipeline.create()`， SageMaker AI 会编译所有定义为管道实例一部分的步骤。 SageMaker AI 将序列化函数、参数和所有其他与步骤相关的项目上传到 Amazon S3。 数据按照以下结构存放在 S3 存储桶中： ``` s3_root_uri/ pipeline_name/ sm_rf_user_ws/ workspace.zip # archive of the current working directory (workdir) step_name/ timestamp/ arguments/ # serialized function arguments function/ # serialized function pre_train_dependencies/ # any dependencies and pre_execution scripts provided for the step execution_id/ step_name/ results # returned output from the serialized function including the model ``` `s3_root_uri`在 SageMaker AI 配置文件中定义，适用于整个管道。如果未定义，则使用默认 SageMaker AI 存储桶。 **注意** 每次 SageMaker AI 编译管道时， SageMaker AI 都会将步骤的序列化函数、参数和依赖项保存在带有当前时间戳的文件夹中。每次运行 `pipeline.create()`、`pipeline.update()`、`pipeline.upsert()` 或 `pipeline.definition()` 时都会出现这种情况。 # 运行管道 以下页面介绍如何使用 Amazon Pipelines 运行 SageMaker 管道，无论是使用 SageMaker AI 资源还是本地运行。 使用该`pipeline.start()`函数开始新的管道运行，就像传统的 A SageMaker I 管道运行一样。有关 `start()` 函数的信息，请参阅 [sagemaker.workflow.pipeline.Pipeline.start](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.pipeline.Pipeline.start)。 **注意** 使用 `@step` 装饰器定义的步骤将作为训练作业运行。因此，请注意以下限制： 账户中的实例限制和训练作业限制。相应更新您的限制，以避免任何节流或资源限制问题。 管道中每运行一个训练步骤的相关货币成本。有关更多详情，请参阅 [Amazon SageMaker 定价](https://www.amazonaws.cn/sagemaker/pricing/)。 ## 从本地运行的管道中读取结果 要查看管道运行的任何步骤的结果，请使用 [execution.result()](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.pipeline._PipelineExecution.result )，如以下代码段所示： ``` execution = pipeline.start() execution.result(step_name="train") ``` **注意** Pipelines 在本地模式下不支持 `execution.result()`。 每次只能检索一个步骤的结果。如果步骤名称由 SageMaker AI 生成，则可以通过以下方式调用`list_steps`来检索步骤名称： ``` execution.list_step() ``` ## 在本地运行管道 您可以像运行传统管道步骤一样，在本地运行带有 `@step` 装饰步骤的管道。有关本地模式管道运行的详细信息，请参阅 [使用本地模式运行管道](pipelines-local-mode.md)。要使用本地模式，请在管道定义中使用 `LocalPipelineSession` 代替 `SageMakerSession`，如下例所示： ``` from sagemaker.workflow.function_step import step from sagemaker.workflow.pipeline import Pipeline from sagemaker.workflow.pipeline_context import LocalPipelineSession @step def train(): training_data = s3.download(....) ... return trained_model step_train_result = train() local_pipeline_session = LocalPipelineSession() local_pipeline = Pipeline( name="", steps=[step_train_result], sagemaker_session=local_pipeline_session # needed for local mode ) local_pipeline.create(role_arn="role_arn") # pipeline runs locally execution = local_pipeline.start() ``` # 配置您的管道 建议您使用 SageMaker AI 配置文件来设置管道的默认值。有关 SageMaker AI 配置文件的信息，请参阅[在 SageMaker Python SDK 中配置和使用默认值](https://sagemaker.readthedocs.io/en/stable/overview.html#configuring-and-using-defaults-with-the-sagemaker-python-sdk)。添加到配置文件中的任何配置都适用于管道中的所有步骤。如果您要覆盖任何步骤的选项，请在 `@step` 装饰器参数中提供新值。下面的主题介绍了如何设置配置文件。 配置文件中 `@remote` 装饰器的配置与 `@step` 装饰器的配置完全相同。要在配置文件中设置管道角色 ARN 和管道标签，请使用以下代码段中的 `Pipeline` 部分： ``` SchemaVersion: '1.0' SageMaker: Pipeline: RoleArn: 'arn:aws:iam::555555555555:role/IMRole' Tags: - Key: 'tag_key' Value: 'tag_value' ``` 在配置文件中设置的大部分默认值，都可以通过向 `@step` 装饰器传递新值来覆盖。例如，您可以覆盖配置文件中为预处理步骤设置的实例类型，如下例所示： ``` @step(instance_type="ml.m5.large") def preprocess(raw_data): df = pandas.read_csv(raw_data) ... return procesed_dataframe ``` 一些参数不是`@step`装饰器参数列表的一部分，只能通过 SageMaker AI 配置文件为整个管道配置这些参数。它们列举如下： + `sagemaker_session`(`sagemaker.session.Session`): SageMaker AI 委托服务调用的底层 SageMaker AI 会话。如果未指定，则使用默认配置创建会话，如下所示： ``` SageMaker: PythonSDK: Modules: Session: DefaultS3Bucket: 'default_s3_bucket' DefaultS3ObjectKeyPrefix: 'key_prefix' ``` + `custom_file_filter`（`CustomFileFilter)`：`CustomFileFilter` 对象，用于指定要包含在管道步骤中的本地目录和文件。如果未指定，默认为 `None`。要使 `custom_file_filter` 生效，您必须将 `IncludeLocalWorkdir` 设置为 `True`。下面的示例显示了忽略所有笔记本文件以及名为 `data` 的文件和目录的配置。 ``` SchemaVersion: '1.0' SageMaker: PythonSDK: Modules: RemoteFunction: IncludeLocalWorkDir: true CustomFileFilter: IgnoreNamePatterns: # files or directories to ignore - "*.ipynb" # all notebook files - "data" # folder or file named "data" ``` 有关如何使用 `IncludeLocalWorkdir` 和 `CustomFileFilter` 的更多详情，请参阅 [将模块化代码用于 @remote 装饰器](train-remote-decorator-modular.md)。 + `s3_root_uri (str)`：Amazon S3 根文件夹， SageMaker AI 将代码档案和数据上传到该文件夹。如果未指定，则使用默认 SageMaker AI 存储桶。 + `s3_kms_key (str)`：用于加密输入和输出数据的键。您只能在 SageMaker AI 配置文件中配置此参数，并且该参数适用于管道中定义的所有步骤。如果未指定，默认为 `None`。请参阅下面的 S3 KMS 密钥配置示例片段： ``` SchemaVersion: '1.0' SageMaker: PythonSDK: Modules: RemoteFunction: S3KmsKeyId: 's3kmskeyid' S3RootUri: 's3://amzn-s3-demo-bucket/my-project ``` # 最佳实践 以下各节建议了在管道步骤中使用 `@step` 装饰器时应遵循的最佳实践。 ## 使用暖池 要加快管道步骤运行速度，可使用为训练作业提供的热池功能。您可以通过为 `@step` 装饰器提供 `keep_alive_period_in_seconds` 参数来开启暖池功能，如以下代码段所示： ``` @step( keep_alive_period_in_seconds=900 ) ``` 有关暖池的更多信息，请参阅 [SageMaker AI 管理的暖池](train-warm-pools.md)。 ## 目录结构 建议您在使用 `@step` 装饰器时使用代码模块。将调用步骤函数和定义管道的 `pipeline.py` 模块放在工作区的根部。建议的结构如下： ``` . ├── config.yaml # the configuration file that define the infra settings ├── requirements.txt # dependencies ├── pipeline.py # invoke @step-decorated functions and define the pipeline here ├── steps/ | ├── processing.py | ├── train.py ├── data/ ├── test/ ``` # 限制 以下各节概述了在管道步骤中使用 `@step` 装饰器时应注意的限制。 ## 函数参数限制 向 `@step` 装饰函数传递输入参数时，会受到以下限制： + 您可以将 `DelayedReturn`、`Properties`（其他类型的步骤）、`Parameter` 和 `ExecutionVariable` 对象作为参数传递给 `@step` 装饰函数。但 `@step` 装饰的函数不支持将 `JsonGet` 和 `Join` 对象作为参数。 + 您不能通过 `@step` 函数直接访问管道变量。下面的示例会产生一个错误： ``` param = ParameterInteger(name="", default_value=10) @step def func(): print(param) func() # this raises a SerializationError ``` + 您不能将管道变量嵌套到另一个对象中，并将其传递给 `@step` 函数。下面的示例会产生一个错误： ``` param = ParameterInteger(name="", default_value=10) @step def func(arg): print(arg) func(arg=(param,)) # this raises a SerializationError because param is nested in a tuple ``` + 由于函数的输入和输出是序列化的，因此可以作为函数的输入或输出传递的数据类型受到限制。更多详情，请参阅 [调用远程函数](train-remote-decorator-invocation.md) 中的*数据序列化和反序列化*部分。同样的限制也适用于 `@step` 装饰函数。 + 任何具有 boto 客户端的对象都不能被序列化，因此不能将此类对象作为 `@step` 装饰函数的输入或输出。例如， SageMaker Python SDK 客户端类（例如`Estimator``Predictor`、和）`Processor`无法序列化。 ## 功能导入 应在函数内部而不是外部导入步骤所需的库。如果在全局范围内导入，就有可能在序列化函数时发生导入碰撞。例如，`sklearn.pipeline.Pipeline` 可以被 `sagemaker.workflow.pipeline.Pipeline` 覆盖。 ## 引用函数返回值的子成员 如果引用 `@step` 装饰函数返回值的子成员，则会受到以下限制： + 如果 `DelayedReturn` 对象表示元组、列表或 dict，则可以用 `[]` 引用子成员，如下例所示： ``` delayed_return[0] delayed_return["a_key"] delayed_return[1]["a_key"] ``` + 由于调用函数时无法知道基础元组或列表的确切长度，因此无法解包元组或列表输出。下面的示例会产生一个错误： ``` a, b, c = func() # this raises ValueError ``` + 您不能遍历 `DelayedReturn` 对象。以下示例会引发错误： ``` for item in func(): # this raises a NotImplementedError ``` + 您不能使用“`.`”引用任意子成员。下面的示例会产生一个错误： ``` delayed_return.a_child # raises AttributeError ``` ## 不支持的现有管道功能 您不能使用具有以下管道功能的 `@step` 装饰器： + [Pipeline 步骤缓存](https://docs.amazonaws.cn/sagemaker/latest/dg/pipelines-caching.html) + [Property 文件](https://docs.amazonaws.cn/sagemaker/latest/dg/build-and-manage-propertyfile.html#build-and-manage-propertyfile-property) # 在步骤之间传递数据 使用 Amazon Pipelines 构建 SageMaker 管道时，您可能需要将数据从一个步骤传递到下一个步骤。例如，您可能希望将训练步骤生成的模型构件作为模型评测或部署步骤的输入。您可以使用该功能创建相互依赖的管道步骤，并构建您的 ML 工作流程。 当您需要从管道步骤的输出中获取信息时，可以使用 `JsonGet`。`JsonGet` 可以帮助您从 Amazon S3 或属性文件中提取信息。下文将介绍使用 `JsonGet` 提取步进输出的方法。 ## 使用 Amazon S3 在不同步骤之间传递数据 您可以在 `ConditionStep` 中使用 `JsonGet`，直接从 Amazon S3 获取 JSON 输出。Amazon S3 URI 可以是包含原始字符串、管道运行变量或管道参数的 `Std:Join` 函数。下面的示例展示了如何在 `ConditionStep` 中使用 `JsonGet`： ``` # Example json file in s3 bucket generated by a processing_step { "Output": [5, 10] } cond_lte = ConditionLessThanOrEqualTo( left=JsonGet( step_name="", s3_uri="", json_path="Output[1]" ), right=6.0 ) ``` 如果在条件步骤中使用带有 Amazon S3 路径的 `JsonGet`，则必须在条件步骤和生成 JSON 输出的步骤之间明确添加依赖关系。在下面的示例中，条件步骤的创建依赖于处理步骤： ``` cond_step = ConditionStep( name="", conditions=[cond_lte], if_steps=[fail_step], else_steps=[register_model_step], depends_on=[processing_step], ) ``` ## 使用属性文件在各步骤之间传递数据 使用属性文件存储处理步骤输出中的信息。这在分析处理步骤的结果以决定如何执行条件步骤时特别有用。该`JsonGet`函数处理属性文件，并允许您使用 JsonPath 符号来查询属性 JSON 文件。有关 JsonPath 符号的更多信息，请参阅 [JsonPath repo](https://github.com/json-path/JsonPath)。 要存储属性文件以备日后使用，必须先创建一个具有以下格式的 `PropertyFile` 实例。`path` 参数是保存属性文件的 JSON 文件的名称。任何 `output_name` 都必须与您在处理步骤中定义的 `ProcessingOutput` 的 `output_name` 相匹配。这使属性文件能够捕获步骤中的 `ProcessingOutput`。 ``` from sagemaker.workflow.properties import PropertyFile = PropertyFile( name="", output_name="", path="" ) ``` 创建`ProcessingStep`实例时，添加`property_files`参数以列出 Amazon SageMaker Pipelines 服务必须索引的所有参数文件。这将保存属性文件以备日后使用。 ``` property_files=[] ``` 要在条件步骤中使用属性文件，请将 `property_file` 添加到传递给条件步骤的条件中（如以下示例所示），以便使用 `json_path` 参数查询 JSON 文件中的所需属性。 ``` cond_lte = ConditionLessThanOrEqualTo( left=JsonGet( step_name=step_eval.name, property_file=, json_path="mse" ), right=6.0 ) ``` 有关更深入的示例，请参阅 [Amaz SageMaker on Python 软件开发工具包](https://sagemaker.readthedocs.io/en/stable)中的*[属性文件](https://sagemaker.readthedocs.io/en/stable/amazon_sagemaker_model_building_pipeline.html#property-file)*。 # 缓存管道步骤 在 Amazon Pipelin SageMaker es 中，您可以在重新运行管道时使用步骤缓存来节省时间和资源。当某个步骤的配置和输入相同时，步骤缓存会重复使用该步骤前一次成功运行的输出（而不是重新计算）。这可帮助您在使用相同参数重新运行管道时获得一致的结果。下面的主题将向您介绍如何为管道配置和开启步骤缓存。 使用步骤签名缓存时，Pipelines 会尝试查找当前管道步骤中某些属性值相同的前一次运行。如果发现，Pipelines 会传播上一次运行的输出结果，而不是重新计算步骤。所检查的属性特定于步骤类型，并在 [按管道步骤类型划分的默认缓存键属性](pipelines-default-keys.md) 中列出。 您必须选择步骤缓存 - 默认情况下，它处于关闭状态。开启步骤缓存时，还必须定义超时。此超时定义了之前运行可以持续多久才能继续作为重复使用的候选项。 步骤缓存仅考虑成功的运行 - 它从不重复使用失败的运行。如果超时时间内有多次成功运行，Pipelines 会使用最近一次成功运行的结果。如果在超时时间内没有成功运行，Pipelines 会重新运行该步骤。如果执行程序发现之前的运行符合条件但仍在进行中，则两个步骤都将继续运行，如果成功运行，则会更新缓存。 步骤缓存仅适用于单个管道，因此即使存在步骤签名匹配项，也无法重复使用另一个管道中的步骤。 步骤缓存可用于以下步骤类型： + [Processing](build-and-manage-steps-types.md#step-type-processing) + [训练](build-and-manage-steps-types.md#step-type-training) + [优化](build-and-manage-steps-types.md#step-type-tuning) + [AutoML](build-and-manage-steps-types.md#step-type-automl) + [转换](build-and-manage-steps-types.md#step-type-transform) + [`ClarifyCheck`](build-and-manage-steps-types.md#step-type-clarify-check) + [`QualityCheck`](build-and-manage-steps-types.md#step-type-quality-check) + [EMR](build-and-manage-steps-types.md#step-type-emr) **Topics** + [ # 开启步骤缓存 ](pipelines-caching-enabling.md) + [ # 关闭步骤缓存 ](pipelines-caching-disabling.md) + [ # 按管道步骤类型划分的默认缓存键属性 ](pipelines-default-keys.md) + [ # 缓存数据访问控制 ](pipelines-access-control.md) # 开启步骤缓存 要开启步骤缓存，必须在步骤定义中添加 `CacheConfig` 属性。管道定义文件中的 `CacheConfig` 属性使用以下格式： ``` {     "CacheConfig": {         "Enabled": false,         "ExpireAfter": "