AWS自动化机器学习-六、使用 AWS 步骤函数自动化机器学习过程

在本书的前三章中，我们看到了一个可能影响 ML 用例自动化的基本过程缺陷，即由 ML 从业者生产的产品级模型向应用开发和操作团队的移交。在 [第 4 章]*中，我们研究了如何通过使用 CI/CD 过程方法将 ML 处理合并到 DevOps 过程中来解决这个问题。

虽然该解决方案本质上解决了问题，但您也可以得出结论，就总体所有权而言，开发或平台团队主要负责构建最终解决方案的大部分。例如，您将回忆起我们在前一章中使用的 CI/CD 示例，应用开发团队构建了管道基础，以及将数据处理、模型训练和模型评估任务卸载给 SageMaker 的集成。这些集成要求开发团队对 ML 有一个基本的了解，以及整个 ML 过程。

虽然这个例子确实显示了 ML 从业者以打包的容器映像的形式向这个跨职能团队提供的不仅仅是一个优化的模型，但事实仍然是，大多数解决方案的开发仍然是应用和平台开发人员的责任，因此要求他们自己本质上也是 ML 从业者。

当然，分配的职责不成比例可能是因为并非所有的 ML 从业者本身都是 DevOps 工程师，并且，从组织上来说，并非所有的 ML 从业者都与开发人员和基础设施人员属于同一个团队。

无论哪种方式，我们如何进一步简化 ML 自动化过程，而不必进一步提高开发和 ML 团队的技能，或者改变组织结构？

回答这个问题将是本章的主要焦点，我们将继续建立在前一章中建立的基础上，并继续简化年龄计算器的例子。为此，我们将讨论以下主题:

AWS 步骤函数简介
使用 AWS Step 函数 Data Science SDK for CI/CD
构建 CI/CD 渠道资源

技术要求

本章的技术要求如下:

web 浏览器(为了获得最佳体验，建议您使用 Chrome 或 Firefox)。
访问您在 [第四章]。
接入我们在 [第四章]*。
我们将再次在 AWS 免费层的使用限制内工作，以避免超出不必要的成本。
本章的配套 GitHub 资源库中提供了源代码示例和 Jupyter 笔记本(https://GitHub . com/packt publishing/Automated-Machine-Learning-on-AWS/tree/main/chapter 06)。Cloud9 开发环境中应该已经有代码示例了。如果没有，参考 第四章机器学习持续集成和持续交付(CI/CD)中题为开发应用工件的章节。

介绍 AWS 步骤功能

在 re:Invent 2016 上，AWS 宣布将步骤功能服务作为一种通过创建工作流来编排常见业务流程的方式。一个工作流，也称为作为状态机，本质上是一系列事件驱动的步骤，或状态，表示一个单一的流程单元。通过将这些工作单元链接在一起，我们有效地创建了一个自动化的过程来完成一个总体目标。

在自动化 ML 流程的情况下，我们可以创建一个状态机，将各个步骤链接在一起，以处理训练数据、训练 ML 模型、评估训练模型的性能，甚至将模型部署到生产中。

在 ML 流程中使用 Step 函数，或者在这种情况下自动化任何工作流程的优势在于，我们可以根据每个步骤的结果来重新定向流程。例如，如果工作流中的某个特定步骤失败了，我们可以重试它，或者重定向整个流程以遵循一些替代的流程逻辑。

创建状态机

为了创建整个工作流，我们首先在状态机中创建各个状态。这是通过使用 Amazon States 语言(https://States-Language . net/spec . html)定义这些状态来实现的。状态语言是一种基于 JSON 的模式，您可以通过它将每个状态手动定义为一个 JSON 对象。下面的代码显示了一个使用 States 语言定义状态机的示例:

{
    "Comment": "A simple minimal example of the States language",
    "StartAt": "Hello World",
    "States": {
    "Hello World": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:HelloWorld",
      "End": true
    }
  }
}

注意

这个例子是在 Apache 许可证 2.0 版本下提供的，并且直接来自亚马逊州语言规范的在线副本(https://states-language.net/spec.html)。 图 6.1 显示了工作流的图形表示，或状态机定义，源自 States Language JSON 模式: ![Figure 6.1 – State machine definition ]

图 6.1-状态机定义

正如您从图 6.1 以及 JSON 模式示例中看到的，我们已经定义了一个名为 Hello World 的状态。我们还进一步指定了这种状态是任务状态，它的工作单元是执行一个 AWS Lambda 函数。此外，我们已经定义了工作流，以此任务开始，并在任务执行后结束。

因此，虽然这看起来非常简单，但正如我们将在第七章、中看到的，使用 AWS 步骤函数构建 ML 工作流，当我们开始定义包括重试和失败的状态并合并不同的选择路径时，最终的 JSON 模式定义可能会非常复杂。

解决状态机的复杂性

自从创建 Step Function 服务以来，AWS 已经提供了一些机制来克服与使用 States 语言手动定义状态机相关的复杂性。

比如 2019 年 7 月，AWS 推出了AWS Toolkit for Visual Studio Code(https://aws.amazon.com/visualstudiocode/)。作为该工具包的一部分，AWS 为开发人员提供了从 VS 代码 IDE 中定义、可视化、执行和更新状态机的能力。除了代码完成和状态机验证，开发人员还可以在使用 VS 代码时克服一些与用 States 语言定义状态机相关的复杂性。

此外，在 2021 年 3 月，AWS 引入了使用YAML Ain not Markup Language(YAML)序列化语言而不是 JSON 来定义状态机的能力，因此使得开发者更容易构建状态机，如果 YAML 是他们选择的序列化语言的话。

再者，2021 年 7 月，AWS 公布了 Workflow Studio 。这是一个可视化的工作流设计工具，允许开发人员使用图形设计工具，在 AWS 控制台中，通过简单地将工作流和任务状态拖放到画布上来构建状态机，并使用最少的代码将它们集成，从而使开发人员更容易构建复杂的工作流。

然而，即使这些增加的功能使得定义基于 ML 的工作流变得更加容易，问题仍然存在:谁最终负责定义状态机模式？ 是应用开发团队还是 ML 从业者？

在下一节中，我们将评估如何使用 Step Functions 功能来帮助数据科学家和 ML 从业者进一步简化和自动化 ML 工作流程。

使用 Step 函数 Data Science SDK for CI/CD

2019 年 11 月，AWS 为亚马逊 SageMaker 推出了AWS Step Functions Data Science SDK。该 SDK 允许数据科学和 ML 实践者以编程方式构建阶跃函数工作流，以交付生产级 ML 模型。SDK 设计用于在 Jupyter 笔记本中构建一个过程，以阶跃函数工作流的形式提供可再现的 ML 实验，而不是再现实验本身。

基本上，这意味着 ML 从业者不是探索数据、构建算法、训练模型和评估训练模型的性能，而是构建一个状态机来自动完成这些任务。在此基础上，产生的状态机是以编程方式构造的，而不是用 States 语言规范手动定义。因此，要回答前一节中提出的问题，ML 从业者现在可以拥有定义自动化过程的任务，以产生生产级 ML 模型。

怎么会这样？

前面我们看到了 ML 从业者如何将预包装的容器图像作为工件交付给 CI/CD 流程。这个工件包含了处理数据、训练模型和评估模型性能的各种运行时过程。我们还看到了开发和运营团队如何重新调整 CI/CD 渠道，以纳入典型的 ML 流程。

现在，本着跨职能团队的精神，ML 从业者可以交付一个状态机，作为 CI/CD 管道工件来自动化整个过程。另外，通过使用 Data Science SDK，状态机可以以编程方式定义，而不必提高 ML 从业者团队的技能。另一方面，开发团队不必提升他们的 ML 知识来将 ML 过程合并到 CI/CD 管道中。

为了准确地展示一个敏捷的跨职能团队将如何创建这个解决方案，让我们从头开始重新分解以前使用的 CI/CD 流程。图 6.2 提供了结果重组流程的概述。 ![Figure 6.2 – Re-factored CI/CD process ]

图 6.2–重构的 CI/CD 流程

在图 6.2 中可以看到，与图 4.5 中概述的过程在 [第四章]中略有不同。在重构的过程中，ML 从业者承担了更多重要的角色。虽然 CI 和 CD 阶段基本保持不变，但是 ML 从业者现在负责开发将编排这些过程的工作流资产。

例如，在重构的过程中，ML 从业者不仅负责从 ML 实验中提供最佳的模型工件，现在他们还负责使用 Step 函数 Data Science SDK 构建和测试自动化的 ML 过程。这个自动化的 ML 工作流工件负责预处理训练数据，训练 ML 模型，并评估模型是否准备好用于生产。

然而，在工作流资产能够集成之前，开发工程师还必须构建 CI/CD 管道。下一节将带您了解如何做到这一点。

建立 CI/CD 渠道资源

为了开始重构年龄计算器用例，我们将从开发和运营团队的角度来完成初始设置步骤。我们将使用我们在 [第 4 章] B17649_04_ePub.xhtml##_idTextAnchor061

更新开发环境
创建管道工件存储库
构建管道应用工件
部署 CI/CD 管道

让我们开始吧。

更新开发环境

首先，登录到您到目前为止一直使用的 AWS 帐户，打开 AWS Cloud9 控制台(https://console . AWS . Amazon . com/cloud 9)。在 Your environments 下，点击 Open IDE 按钮，启动 MLOps-IDE 开发环境。

注意

如果您还没有准备好 MLOps-IDE 环境，请参考 [第四章]的准备开发环境*部分。

要更新环境，请执行以下步骤:

Run the following command to ensure that we have version 2.3.0 of the CDK installed:
```
$ cdk --version
```
注意在撰写本文时，CDK 的最新版本是 2.3.0 (build beaa5b2) 。如果您没有在 Cloud9 环境中运行该版本，请参考 [第 4 章]*，了解如何安装它。

既然我们已经将环境更新到了 CDK 的最新版本，我们就可以创建源代码存储库了。

创建管道工件库

执行下面的步骤来创建一个新的abalone-cicd-pipeline代码提交库:

使用工作区终端，运行以下 CLI 命令以确保 CLI 区域设置正确。确保在 CodeCommit 管理控制台(https://console . AWS . Amazon . com/code suite/code commit/repositories)中替换abalone-cicd-pipeline CodeCommit 存储库。

使用以下命令创建名为abalone-cicd-pipeline的新 CodeCommit 存储库:

$ aws codecommit create-repository --repository-name abalone-cicd-pipeline --repository-description "Automated ML on AWS using the Step Functions Data Science SDK"

接下来，捕获新创建的存储库的 URL，以便克隆它。运行下面的命令创建CLONE_URL参数:

$ CLONE_URL=$(aws codecommit get-repository --repository-name abalone-cicd-pipeline --query "repositoryMetadata.cloneUrlHttp" --output text)

运行以下命令在本地克隆空存储库:
```
$ git clone $CLONE_URL
```

现在我们有了新的项目存储库，我们可以继续下一个任务，构建应用工件。

构建管道应用构件

使用以下步骤构建管道应用:

通过运行以下命令初始化一个新的 CDK 项目:

$ cd ~/environment/abalone-cicd-pipeline && cdk init --language python

通过运行以下命令设置源存储库的主分支:

$ git add -A
$ git commit -m "Started Pipeline Project"
$ git branch main
$ git checkout main

通过运行以下命令配置 Python 环境:

$ source .venv/bin/activate
$ python -m pip install -U pip pylint boto3
$ pip install -r requirements.txt

您可能还记得 [第四章]可以收集功能需求，使用以下步骤构建端点:

使用 Cloud9 工作区的导航面板，展开abalone-ci-cd-pipeline文件夹，然后右键单击abalone_cicd_pipeline文件夹并选择新文件选项。
Name the newly created file abalone_endpoint_stack.py and double-click on it for editing. 注意在配套的 GitHub 资源库(https://GitHub . com/packt publishing/Automated-Machine-Learning-on-AWS/tree/main/chapter 06/CDK)中可以查看到abalone_endpoint_stack.py的完整副本。该文件也应该可以在 Cloud9 环境中的~/environment/src/Chapter06/cdk/文件夹中查看。

在 Python 文件中，添加以下代码来导入端点的 CDK 模块:

...
import aws_cdk as cdk
import aws_cdk.aws_sagemaker as sagemaker
...

接下来，通过添加以下代码创建一个名为EndpointStack()的 Python 类作为 CDK 堆栈结构:

...
class EndpointStack(cdk.Stack):
    def __init__(self, app: cdk.App, id: str, *, model_name: str=None, repo_name: str=None, **kwargs) -> None:
        super().__init__(app, id, **kwargs)
...

现在，我们使用下面的代码为 S3 桶和管道执行 ID 创建所需的管道参数。这些参数将在流水线执行期间导出:

...
        bucket_name = cdk.CfnParameter(
            self,
            "BucketName",
            type="String"
        )
        execution_id = cdk.CfnParameter(
            self,
            "ExecutionId",
            type="String"
        )
...

由于我们已经实例化了端点构造的参数，现在可以在下面的代码片段中定义端点配置。该配置详细说明了在哪些计算资源上运行端点，以及为端点托管的经过训练的模型，以及在哪里存储用于模型监控的推理和响应数据。下面的代码展示了如何将endpoint_config变量实例化为CfnEndpointConfig() :

...
        endpoint_config = sagemaker.CfnEndpointConfig(
            self,
            "EndpointConfig",
            endpoint_config_name="{}-config-{}".format(model_name.capitalize(), execution_id.value_as_string),
            production_variants=[
                sagemaker.CfnEndpointConfig.ProductionVariantProperty(
                    initial_instance_count=2,
                    initial_variant_weight=1.0,
                    instance_type="ml.m5.large",
                    model_name="{}-{}".format(model_name, execution_id.value_as_string),
                    variant_name="AllTraffic"
                )
            ],
...

从前面的代码片段继续，我们继续定义CfnEndpointConfig()并指定data_capture_config参数。如下面的代码片段所示，我们指定了一个DataCaptureConfigProperty()，它将端点配置为将 100%的输入数据捕获到端点，并将来自端点的输出数据捕获到 S3:

...
            data_capture_config=sagemaker.CfnEndpointConfig.DataCaptureConfigProperty(
                capture_content_type_header=sagemaker.CfnEndpointConfig.CaptureContentTypeHeaderProperty(
                    csv_content_types=[
                        "text/csv"
                    ]
                ),
                capture_options=[
                    sagemaker.CfnEndpointConfig.CaptureOptionProperty(capture_mode="Input"),
                    sagemaker.CfnEndpointConfig.CaptureOptionProperty(capture_mode="Output")
                ],
                destination_s3_uri="s3://{}/endpoint-data-capture".format(bucket_name.value_as_string),
                enable_capture=True,
                initial_sampling_percentage=100.0
            )
        )
...

构造的最后一个部分是端点本身。我们使用下面的代码来声明endpoint，然后保存文件:

...
        endpoint = sagemaker.CfnEndpoint(
            self,
            "AbaloneEndpoint",
            endpoint_config_name=endpoint_config.attr_endpoint_config_name,
            endpoint_name="-Endpoint".format(model_name.capitalize())
        )
        endpoint.add_depends_on(endpoint_config)
...

既然已经创建了端点部署构造，我们现在需要创建构建脚本，该脚本从正在运行的 CI/CD 管道中捕获执行参数。以下步骤将引导您完成这一过程:

使用 Cloud9 终端，运行以下命令来创建必要的工件文件夹:
```
$ mkdir -p ~/environment/abalone-cicd-pipeline/artifacts/scripts
```
使用导航面板，右击新创建的scripts文件夹并选择新文件选项。
Name the file deploy.py and double-click on it for editing. 注意在配套的 GitHub 资源库(https://GitHub . com/packt publishing/Automated-Machine-Learning-on-AWS/tree/main/chapter 06/scripts)中可以查看到deploy.py的完整副本。该文件也应该可以在 Cloud9 环境中的~/environment/src/Chapter06/scripts/文件夹中查看。
在deploy.py文件中，我们要做的第一件事是添加下面的代码片段到导入我们将要使用的 Python 库:
```
...
import boto3
import logging
import os
import json
import sys
from botocore.exceptions import ClientError
...
```

接下来，我们添加以下代码片段来指定全局参数，例如日志记录和用于 CodePipeline 和 SageMaker 的 AWS Python SDK ( boto3)客户端:

...
logger = logging.getLogger()
logging_format = "%(levelname)s: [%(filename)s:%(lineno)s] %(message)s"
logging.basicConfig(format=logging_format, level=os.environ.get("LOGLEVEL", "INFO").upper())
codepipeline_client = boto3.client("codepipeline")
sagemaker_client = boto3.client("sagemaker")
pipeline_name = os.environ["PIPELINE_NAME"]
model_name = os.environ["MODEL_NAME"]
...

下面的代码片段显示了一个名为get_execution_id()的函数。这个函数调用正在运行的 CI/CD 管道，并返回当前执行的 ID。这个 ID 用于版本化将作为 SageMaker 端点托管的模型:

...
def get_execution_id(name=None, task=None):
    try:
        response = codepipeline_client.get_pipeline_state(name=name)
        for stage in response["stageStates"]:
            if stage["stageName"] == "Deploy":
                for action in stage["actionStates"]:
                    if action["actionName"] == task:
                        return stage["latestExecution"]["pipelineExecutionId"]
    except ClientError as e:
        error = e.response["Error"]["Message"]
        logger.error(error)
        raise Exception(error)
...

最后，下面的代码片段实例化了主程序。这个程序创建了,并将执行 ID 和 S3 桶名存储在一个params.json文件中。这个 JSON 文件将被用作端点部署堆栈的输入参数:

...
if __name__ == "__main__":
    task = "DeploymentBuild"
    execution_id = get_execution_id(name=pipeline_name, task=task)
    logger.info("Creating Stack Parameters")
    params = {
        "ExecutionId": execution_id,
        "BucketName": os.environ["BUCKET_NAME"]
    }
    try:
        with open(os.path.join(os.environ["CODEBUILD_SRC_DIR"], "output/params.json"), "w") as f:
            json.dump(params, f)
        logger.info(json.dumps(params, indent=4)),
        sys.exit(0)
    except Exception as error:
        logger.error(error)
        sys.exit(255)
...

在deploy.py文件中输入之前的代码后，确保保存。

至此，我们已经创建了必要的 CDK 构造和支持部署代码，用于将模型作为 SageMaker 端点托管。现在，我们可以通过以下步骤继续创建 CI/CD 管道结构:

Within the Cloud9 navigation panel, expand the abalone_cicd_pipeline folder and double-click on the abalone_cicd_pipeline_stack.py file for editing. 注意在配套的 GitHub 资源库(https://GitHub . com/packt publishing/Automated-Machine-Learning-on-AWS/tree/main/chapter 06/scripts)中可以查看到abalone_cicd_pipeline_stack.py文件的完整副本。该文件还应该可以从 Cloud9 环境中的~/environment/src/Chapter06/cdk/文件夹复制到abalone_cicd_pipeline文件夹中。

如果您选择从头开始创建文件，请删除文件中的任何现有模板代码，以便我们可以从一个空白文件开始，并添加以下代码来导入管道构造所需的 CDK 模块:

...
import os
import aws_cdk.core as cdk
import aws_cdk.aws_codecommit as codecommit
import aws_cdk.aws_codepipeline as codepipeline
import aws_cdk.aws_codepipeline_actions as pipeline_actions
import aws_cdk.aws_codebuild as codebuild
import aws_cdk.aws_iam as iam
import aws_cdk.aws_ecr as ecr
import aws_cdk.aws_s3 as s3
import aws_cdk.aws_s3_deployment as s3_deployment
import aws_cdk.aws_ssm as ssm
from constructs import Construct
...

接下来，我们添加以下代码片段来创建一个名为PipelineStack()的新 Python 类作为 CDK 堆栈结构:

...
class PipelineStack(cdk.Stack):
    def __init__(self, scope: Construct, id: str, *, model_name: str=None, repo_name: str=None, cdk_version: str=None, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)
...

我们为 CDK 构造定义的第一个组件是 CodeCommit 存储库的占位符。正如您在下面的代码片段中看到的，我们使用from_repository_name方法引用了之前创建的abalone-cicd-pipeline CodeCommit 存储库，并将我们的存储库变量引用为repo_name :
```
...
        code_repo = codecommit.Repository.from_repository_name(
            self,
            "PipelineSourceRepo",
            repository_name=repo_name
        )
...
```

下一个要创建的组件是 IAM 策略文档。IAM 角色将使用这个策略，不仅执行状态机，还访问工作流中使用的各种 AWS 资源。我们定义了一个名为workflow_policy_document的变量，并创建了一个 Python 字典来存储各种 IAM 策略语句。下面的代码片段显示了 IAM 策略的Action语句的摘录。在这里，您可以看到我们给了 IAM 角色访问权，以获取当前的 CI/CD 管道执行，并调用状态机本身包含的任何 Lambda 函数。我们还通过为角色提供创建、删除、描述和启动任何状态机的能力，赋予角色管理状态机的能力:

...
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "codepipeline:GetPipelineState",
                        "lambda:InvokeFunction",
                        "lambda:UpdateFunctionCode",
                        "lambda:CreateFunction",
                        "states:CreateStateMachine",
                        "states:UpdateStateMachine",
                        "states:DeleteStateMachine",
                        "states:DescribeStateMachine",
                        "states:StartExecution"
                    ],
                    "Resource": "*"
                },
...

grant-least-privilege).

6.  现在我们已经定义了策略语句，我们可以通过声明`workflow_role`变量:

...
        workflow_role = iam.Role(
            self,
            "WorkflowExecutionRole",
            assumed_by=iam.CompositePrincipal(
                iam.ServicePrincipal("codebuild.amazonaws.com")
            )
        )
...

    来创建使用该策略的工作流执行角色
7.  下面的代码片段展示了我们如何使用`attach_inline_policy()`方法将策略文档作为内联策略应用于`workflow_role`:

...
        workflow_role.attach_inline_policy(
            iam.Policy(
                self,
                "WorkflowRoleInlinePolicy",
                document=iam.PolicyDocument.from_json(workflow_policy_document)
            )
        )
...

8.  因为状态机中的任何 Lambda 函数，以及状态机本身，都需要一个 AWS 服务角色，所以我们将这两个服务主体添加到`workflow_role`中，赋予角色承担服务角色的权限。下面的代码片段展示了我们如何使用`add_statements()`方法来提供`AssumeRole`功能:

...
        workflow_role.assume_role_policy.add_statements(
            iam.PolicyStatement(
                actions=[
                    "sts:AssumeRole"
                ],
                effect=iam.Effect.ALLOW,
                principals=[
                    iam.ServicePrincipal("lambda.amazonaws.com"),
                    iam.ServicePrincipal("sagemaker.amazonaws.com"),
                    iam.ServicePrincipal("states.amazonaws.com")
                ]
            )
        )
...

9.  由于工作流将执行各种 SageMaker 函数来训练、评估和托管 ML 模型，下面的代码片段显示了我们如何通过使用`add_managed_policy()`方法向`workflow_role`提供`AmazonSageMakerFullAccess`托管 IAM 策略:

...
        workflow_role.add_managed_policy(
            iam.ManagedPolicy.from_aws_managed_policy_name("AmazonSageMakerFullAccess")
        )
...

10.  现在我们添加这个**亚马逊资源名** ( **ARN** )作为参数。该参数将由 ML 从业者在定义工作流程时使用，并将存储在 **AWS 系统管理器参数存储库** ( **SSM** )中。这样，当 ML 从业者团队需要引用角色时，他们可以通过使用 API 调用参数存储:

...
        workflow_role_param = ssm.StringParameter(
            self,
            "WorkflowRoleParameter",
            description="Step Functions Workflow Execution Role ARN",
            parameter_name="WorkflowRoleParameter",
            string_value=workflow_role.role_arn
        )
        workflow_role_param.grant_read(workflow_role)
...

11.  接下来，我们定义一个 SageMaker 执行角色。SageMaker 将使用这个角色来访问处理训练数据、训练模型和评估模型所需的各种服务:

...
        sagemaker_role = iam.Role(
            self,
            "SageMakerBuildRole",
            assumed_by=iam.CompositePrincipal(
                iam.ServicePrincipal("sagemaker.amazonaws.com")
            ),
            managed_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name("AmazonSageMakerFullAccess")
            ]
        )
…

12.  下面的代码片段显示了如何创建一个 S3 存储桶来存储在 CI/CD 管道执行期间创建的各种资产:

…
        s3_bucket = s3.Bucket(
            self,
            "PipelineBucket",
            bucket_name=f"{repo_name}-{cdk.Aws.REGION}-{cdk.Aws.ACCOUNT_ID}",
            removal_policy=cdk.RemovalPolicy.DESTROY,
            versioned=True
        )
        s3_bucket.grant_read_write(sagemaker_role)
        s3_bucket.grant_read_write(workflow_role)
…

13.  正如 SageMaker 角色的情况一样，我们还需要向 ML 从业者提供对 S3 桶名称的访问，以便在管道之外使用。下面的代码片段显示了我们如何在 SSM 参数存储中存储 S3 存储桶的名称:

…
        s3_bucket_param = ssm.StringParameter(
            self,
            "PipelineBucketParameter",
            description="Pipeline Bucket Name",
            parameter_name="PipelineBucketName",
            string_value=s3_bucket.bucket_name
        )
…

14.  现在已经创建了 S3 存储桶，我们可以使用`BucketDeployment()`方法:

…
        s3_deployment.BucketDeployment(
            self,
            "DeployData",
            sources=[
                s3_deployment.Source.asset(os.path.join(os.path.dirname(__file__), '../artifacts/data'))
            ],
            destination_bucket=s3_bucket,
            destination_key_prefix="abalone_data/raw",
            retain_on_delete=False
        )
…

    复制原始训练数据集
15.  你会回忆起 [*第四章*]*，我们创建了多个 CodeBuild 项目。每个项目都与 ML 流程中的特定任务相关联。由于 ML 从业者将会把整个 ML 过程自动化为一个 Step Functions 状态机，我们现在只需要定义一个单独的 CodeBuild 项目来构建状态机工件作为管道资产。下面的代码片段显示了我们如何定义这个项目，并将其实例化为`workflow_build`变量:

...
        workflow_build = codebuild.Project(
            self,
            "WorkflowBuildProject",
            project_name="WorkflowBuildProject",
            description="CodeBuild Project for Building and Executing the ML Workflow",
            role=workflow_role,
            source=codebuild.Source.code_commit(
                repository=code_repo
            ),
            environment=codebuild.BuildEnvironment(
                build_image=codebuild.LinuxBuildImage.STANDARD_5_0
            ),
            environment_variables={
                "PIPELINE_NAME": codebuild.BuildEnvironmentVariable(
                    value=repo_name
                ),
                "MODEL_NAME": codebuild.BuildEnvironmentVariable(
                    value=model_name
                ),
                "BUCKET_NAME": codebuild.BuildEnvironmentVariable(
                    value=s3_bucket.bucket_name
                )
            }
        )
...

16.  我们还需要创建一个额外的 CodeBuild 项目，通过执行我们之前创建的`deploy.py`脚本来构建模型部署参数。下面的代码片段展示了如何创建附加的 CodeBuild 项目，并通过声明`deployment_build`参数:

...
        deployment_build = codebuild.PipelineProject(
            self,
            "DeploymentBuild",
            project_name="DeploymentBuild",
            description="CodeBuild Project to Synthesize a SageMaker Endpoint CloudFormation Template",
            environment=codebuild.BuildEnvironment(
                build_image=codebuild.LinuxBuildImage.STANDARD_5_0
            ),
            environment_variables={
                "BUCKET_NAME": codebuild.BuildEnvironmentVariable(
                    value=s3_bucket.bucket_name
                ),
                "PIPELINE_NAME": codebuild.BuildEnvironmentVariable(
                    value=repo_name
                ),
                "MODEL_NAME": codebuild.BuildEnvironmentVariable(
                    value=model_name
                )
            },
...

    来实例化它
17.  `deployment_build`的构建规范或构建指令有三个阶段，即`install`、`build`和`post_build`。下面的代码片段显示了`install`阶段，在这里我们提供命令来安装 AWS CDK 和创建端点部署构造所需的相关 Python 库:

...
                        "install": {
                            "runtime-versions": {
                                "python": 3.8,
                                "nodejs": 12
                            },
                            "commands": [
                                "echo Updating build environment",
                                "npm install aws-cdk@{}".format(cdk_version),
                                "python -m pip install --upgrade pip",
                                "python -m pip install -r requirements.txt"
                            ]
                        },
...

18.  在安装了相关的库之后，在`install`阶段，我们合成端点部署构造，并以 JSON 格式创建结果 CloudFormation 模板的输出，称为`EndpointStack.template.json`。下面的代码片段显示了在`build`阶段使用的构建命令:

...
                        "build": {
                            "commands": [
                                "echo Synthesizing cdk template",
                                "npx cdk synth -o output"
                            ]
                        },
...

19.  一旦合成了 CloudFormation 模板，构建规范的最后阶段就是执行`deploy.py`。您将回忆起前面的步骤，`deploy.py`文件创建了`params.json`文件来存储当前的 CI/CD 管道执行参数。下面的代码片段展示了`post_build`阶段的一个例子:

...
                        "post_build": {
                            "commands": [
                                "python ./artifacts/scripts/deploy.py"
                            ]
                        }
                    },
...

20.  既然已经定义了 CI/CD 管道所需的所有组件和工件，我们最终可以定义 CI/CD 管道本身了。然而，在我们定义管道之前，我们需要定义输出工件变量。这些变量对应于初始源代码，以及由`deployment_build`创建的各种输出文件。下面的代码片段显示了被声明的`main_source_output`、`model_source_output`和`deployment_build_output`变量:

...
        main_source_output = codepipeline.Artifact()
        model_source_output = codepipeline.Artifact()
        deployment_build_output = codepipeline.Artifact("DeploymentBuildOutput")
...

21.  既然我们已经声明了各种源工件，我们可以继续使用来自`codepipeline` CDK 结节的`Pipeline()`方法来定义 CI/CD 管道。管道有四个阶段，即源阶段、构建阶段、批准阶段和部署阶段。下面的代码片段定义了源阶段。正如你所看到的，在这个阶段，我们声明了 CodeCommit 库的两个分支，一个用于管道 CDK 代码(T2 分支)，一个用于模型工作流工件(T3 分支):

...
                    stage_name="Source",
                    actions=[
                        pipeline_actions.CodeCommitSourceAction(
                            action_name="MainSource",
                            branch="main",
                            repository=code_repo,
                            output=main_source_output
                        ),
                        pipeline_actions.CodeCommitSourceAction(
                            action_name="ModelSource",
                            branch="model",
                            repository=code_repo,
                            output=model_source_output
                        )
                    ]
                ),
...

22.  在构建阶段，我们定义了一个调用`workflow_build` CodeBuild 项目的动作。您会记得这是创建和执行 Step Functions 状态机的 CodeBuild 项目。下面的代码片段声明了引用`workflow_build`项目的`BuildModel`阶段动作:

...
                codepipeline.StageProps(
                    stage_name="Build",
                    actions=[
                        pipeline_actions.CodeBuildAction(
                            action_name="BuildModel",
                            project=workflow_build,
                            input=model_source_output,
                            run_order=1
                        )
                    ]
                ),
...

23.  CI/CD 管道的倒数第二个阶段是批准阶段。正如您在下面的代码片段中看到的，这里我们创建了一个名为`EvaluationApproval`的阶段操作，通过它我们使用来自`pipeline_actions` CDK 模块的`ManualApprovalAction()`方法向管道添加一个手动批准步骤。在管道执行的这一点上，流程所有者将验证模型已经准备好部署到生产中:

                codepipeline.StageProps(
                    stage_name="Approval",
                    actions=[
                        pipeline_actions.ManualApprovalAction(
                            action_name="EvaluationApproval",
                            additional_information="Is the Model Ready for Production?"
                        )
                    ]
                ),
...

24.  管道的最后阶段是将经过训练的模型部署为 SageMaker 托管的端点。正是在这个阶段，使用`DeployEndpoint`阶段动作部署了先前定义的`EndpointStack()`构造。然而，正如您在下面的代码片段中看到的，在可以使用 CDK 构造之前，我们定义了一个名为`DeploymentBuild`的阶段动作，由此我们运行`deployment_build` CodeBuild 项目来合成执行`CloudFormationCreateUpdateStackAction()` :

...
                codepipeline.StageProps(
                    stage_name="Deploy",
                    actions=[
                        pipeline_actions.CodeBuildAction(
                            action_name="DeploymentBuild",
                            project=deployment_build,
                            input=main_source_output,
                            outputs=[deployment_build_output],
                            run_order=1
                        ),
                        pipeline_actions.CloudFormationCreateUpdateStackAction(
                            action_name="DeployEndpoint",
                            stack_name="EndpointStack",
                            template_path=deployment_build_output.at_path(
                                "EndpointStack.template.json"
                            ),
                            admin_permissions=True,
                            parameter_overrides={
                                "ExecutionId": deployment_build_output.get_param("params.json", "ExecutionId"),
                                "BucketName": deployment_build_output.get_param("params.json", "BucketName"),
                            },
                            extra_inputs=[deployment_build_output],
                            run_order=2
                        )
                    ]
                )
            ]
        )
...

    所需的 CloudFormation 模板以及 CloudFormation 参数文件

既然已经定义了所有的 CDK 构造，我们可以把它们放在一起，定义 CDK 应用。以下步骤将向您展示如何做到这一点:
1.  在`abalone-cicd-pipeline`文件夹中，打开`app.py`文件进行编辑。
2.  删除现有的模板代码，并添加以下代码来定义 CDK 应用:

!/usr/bin/env python3

import os
from aws_cdk import core as cdk
from abalone_cicd_pipeline.abalone_endpoint_stack import EndpointStack
from abalone_cicd_pipeline.abalone_cicd_pipeline_stack import PipelineStack
MODEL = "abalone"
CODECOMMIT_REPOSITORY = "abalone-cicd-pipeline"
CDK_VERSION = "2.3.0"
app = cdk.App()
EndpointStack(
    app,
    "EndpointStack",
    env=cdk.Environment(account=os.getenv("CDK_DEFAULT_ACCOUNT"), region=os.getenv("CDK_DEFAULT_REGION")),
    model_name=MODEL,
    repo_name=CODECOMMIT_REPOSITORY
)
PipelineStack(
    app,
    CODECOMMIT_REPOSITORY,
    env=cdk.Environment(account=os.getenv("CDK_DEFAULT_ACCOUNT"), region=os.getenv("CDK_DEFAULT_REGION")),
    model_name=MODEL,
    repo_name=CODECOMMIT_REPOSITORY,
    cdk_version=CDK_VERSION
)
app.synth()

3.  保存并关闭`app.py`文件。

既然 CI/CD 管道已经完成，我们就可以开始部署它了。

## 部署 CI/CD 管道

执行下面的命令来部署 CDK 应用并创建 CI/CD 管道:
1.  使用 Cloud9 工作区的终端窗口，运行以下命令从 UCI 存储库中下载鲍鱼训练数据:

$ cd ~/environment/abalone-cicd-pipeline/ && mkdir -p artifacts/data
$ wget -c -P artifacts/data https://archive.ics.uci.edu/ml/machine-learning-databases/abalone/abalone.data

2.  为了确保我们刚刚创建的所有代码都提交到 CodeCommit 存储库中，运行以下命令来更新这些更改:

$ git add -A
$ git commit -m "Initial commit of Pipeline Artifacts"
$ git push --set-upstream origin main

3.  现在，运行以下命令来部署 CDK 应用:

$ cdk deploy abalone-cicd-pipeline

正如我们在前一章的中看到的，您可以通过点击控制台导航窗格中的**管道**来查看代码管道控制台(https://console.aws.amazon.com/codesuite/codepipeline/)中的管道。*图 6.3* 显示了您可能会看到的一个例子:
![Figure 6.3 – CodePipeline console
]

图 6.3–代码管道控制台

从*图 6.3* 中可以看到，**鲍鱼-cicd-管道**出现故障。如果你点击管道打开细节，你会看到管道失败是因为没有**模型源**。这是,因为 ML 从业者团队还没有创建任何模型源工件。在下一章中，我们将使用数据科学 SDK 创建状态机工件。

## 总结

在这一章中，我们重新分解了来自 [*第 4 章*],通过基于开发团队和 ML 从业者团队的专业领域集成他们来进一步简化整个 ML 过程。

例如，有了这个重构的过程，开发团队现在可以将他们的专业知识集中在构建和开发 CI/CD 组件上，而 ML 从业者团队可以通过使用 Data Science SDK 专注于编写 ML 过程。

在下一章中，我们将把角色切换到 ML 从业者团队，并回顾他们如何将 ML 工作流编码为步骤功能状态机。

## 文章列表
- [AWS自动化机器学习-一、AWS 上的自动化机器学习入门](https://www.oomspot.com/post/shangdezidonghuajiqixuexirumen)
- [AWS自动化机器学习-七、使用 AWS 步骤函数构建 ML 工作流](https://www.oomspot.com/post/uzhouhanshugoujianmlgongzuoliu)
- [AWS自动化机器学习-三、使用 AutoGluon 技术自动化复杂的模型开发](https://www.oomspot.com/post/ishuzidonghuafuzademoxingkaifa)
- [AWS自动化机器学习-九、使用 Amazon Managed Workflows 为 Apache AirFlow 构建 ML 工作流](https://www.oomspot.com/post/acheairflowgoujianmlgongzuoliu)
- [AWS自动化机器学习-二、使用 SageMaker 自动驾驶器自动化机器学习模型开发](https://www.oomspot.com/post/izidonghuajiqixueximoxingkaifa)
- [AWS自动化机器学习-五、自动化 ML 模型的持续部署](https://www.oomspot.com/post/uzidonghuamlmoxingdechixubushu)
- [AWS自动化机器学习-八、使用 Apache Airflow 实现机器学习过程的自动化](https://www.oomspot.com/post/anjiqixuexiguochengdezidonghua)
- [AWS自动化机器学习-六、使用 AWS 步骤函数自动化机器学习过程](https://www.oomspot.com/post/nshuzidonghuajiqixuexiguocheng)
- [AWS自动化机器学习-十、机器学习软件开发生命周期（MLSDLC）简介](https://www.oomspot.com/post/fashengmingzhouqimlsdlcjianjie)
- [AWS自动化机器学习-十一、MLSDLC 的持续集成、部署和训练](https://www.oomspot.com/post/lcdechixujichengbushuhexunlian)
- [AWS自动化机器学习-四、机器学习的持续集成和持续交（CI/CD）](https://www.oomspot.com/post/idechixujichenghechixujiaocicd)
- [AWS自动化机器学习-第一部分：自动机器学习过程和 AWS 上的 AutoML 的基础知识](https://www.oomspot.com/post/eawsshangdeautomldejichuzhishi)
- [AWS自动化机器学习-第三部分：优化以源代码为中心的自动化机器学习方法](https://www.oomspot.com/post/gxindezidonghuajiqixuexifangfa)
- [AWS自动化机器学习-第二部分：通过持续集成和持续交付（CI/CD）实现机器学习过程的自动化](https://www.oomspot.com/post/anjiqixuexiguochengdezidonghua)
- [AWS自动化机器学习-第五部分：自动化 AWS 上的端到端生产应用](https://www.oomspot.com/post/deduandaoduanshengchanyingyong)
- [AWS自动化机器学习-第四部分：优化以数据为中心的自动化机器学习方法](https://www.oomspot.com/post/gxindezidonghuajiqixuexifangfa)
- [AWS自动化机器学习-零、前言](https://www.oomspot.com/post/szidonghuajiqixuexilingqianyan)

AWS自动化机器学习-六、使用 AWS 步骤函数自动化机器学习过程

技术要求

介绍 AWS 步骤功能

创建状态机

解决状态机的复杂性

使用 Step 函数 Data Science SDK for CI/CD

建立 CI/CD 渠道资源

更新开发环境

创建管道工件库

构建管道应用构件

grant-least-privilege).

!/usr/bin/env python3

更多推荐

更多

Apache CN

近期文章

更多

AWS自动化机器学习-第一部分：自动机器学习过程和 AWS 上的 AutoML 的基础知识

AWS自动化机器学习-第四部分：优化以数据为中心的自动化机器学习方法

AWS自动化机器学习-第二部分：通过持续集成和持续交付（CI/CD）实现机器学习过程的自动化

AWS自动化机器学习-十、机器学习软件开发生命周期（MLSDLC）简介

AWS自动化机器学习-三、使用 AutoGluon 技术自动化复杂的模型开发

AWS自动化机器学习-第五部分：自动化 AWS 上的端到端生产应用

热门标签

更多

推荐作者

更多