在多云环境中使用亚马逊SageMaker训练和部署机器学习模型

随着客户加速迁移到云端并转变业务，有些客户发现自己需要在多云环境中管理IT运营。例如，您可能已经收购了一个已经在不同云服务提供商上运行的公司，或者您可能有一个通过AWS提供的独特功能产生价值的工作负载。另一个例子是独立软件供应商（ISV）将其产品和服务提供给不同的云平台，以满足其最终客户的需求。或者，一个组织可能在一个主要云服务提供商不可用的区域运营，并且为了满足数据主权或数据居住地要求，他们可以使用辅助云服务提供商。

在这些情况下，当您开始将生成式人工智能（AI）、大型语言模型（LLM）和机器学习（ML）技术作为业务的核心部分时，您可能正在寻找在多云环境中利用AWS AI和ML能力的选项。例如，您可能希望利用Amazon SageMaker构建和训练ML模型，或者使用Amazon SageMaker Jumpstart部署预构建的基础或第三方ML模型，您可以仅需点击几下即可部署。或者，您可能希望利用Amazon Bedrock构建和扩展生成式AI应用程序，或者利用AWS的预训练AI服务，这不需要您学习机器学习技能。AWS支持组织将他们自己的模型带到Amazon SageMaker或Amazon SageMaker Canvas中进行预测的场景。

在本文中，我们展示了在多云环境中利用AWS最广泛和最深入的AI/ML能力的众多选项之一。我们展示了如何在AWS中构建和训练ML模型，并在另一个平台上部署该模型。我们使用Amazon SageMaker训练模型，将模型工件存储在Amazon Simple Storage Service（Amazon S3）中，并在Azure上部署和运行模型。如果您在我们讨论的情况之一中使用AWS服务进行ML的最全面功能集，同时需要在其他云服务提供商上运行模型，这种方法将非常有益。

关键概念

Amazon SageMaker Studio是一个基于Web的综合开发环境（IDE），用于机器学习。SageMaker Studio允许数据科学家、ML工程师和数据工程师在一个Web界面上准备数据、构建、训练和部署ML模型。使用SageMaker Studio，您可以访问针对ML开发生命周期的每个阶段的专用工具，从数据准备到构建、训练和部署您的ML模型，将数据科学团队的生产力提高到十倍。SageMaker Studio笔记本是快速启动、协作性的笔记本，与SageMaker和其他AWS服务中的专用ML工具集成。

SageMaker是一个全面的ML服务，使业务分析师、数据科学家和MLOps工程师能够为任何用例构建、训练和部署ML模型，而不受ML专业知识的限制。

AWS为流行的ML框架（如PyTorch、TensorFlow和Apache MXNet）提供了深度学习容器（DLC），您可以将其与SageMaker一起用于训练和推理。DLC以Docker镜像的形式在Amazon Elastic Container Registry（Amazon ECR）中提供。这些Docker镜像已预先安装并经过测试，包括最新版本的流行深度学习框架以及用于训练和推理的其他依赖项。有关由SageMaker管理的预构建Docker镜像的完整列表，请参阅Docker Registry Paths and Example Code。Amazon ECR支持安全扫描，并与Amazon Inspector漏洞管理服务集成，以满足您组织的镜像合规安全要求，并自动化漏洞评估扫描。组织还可以使用AWS Trainium和AWS Inferentia，以获得更好的性价比来运行ML训练作业或推理。

解决方案概述

在本节中，我们将介绍如何使用SageMaker构建和训练模型，并将模型部署到Azure Functions。我们使用SageMaker Studio笔记本来构建、训练和部署模型。我们使用SageMaker中为PyTorch准备的预构建Docker镜像来训练模型。虽然在这种情况下我们将训练好的模型部署到Azure，但您可以使用相同的方法将模型部署到其他平台，如本地或其他云平台。

当我们创建训练作业时，SageMaker会启动ML计算实例，并使用我们的训练代码和训练数据集来训练模型。它将结果模型工件和其他输出保存在我们指定为训练作业输入的S3存储桶中。当模型训练完成后，我们使用Open Neural Network Exchange（ONNX）运行时库将PyTorch模型导出为ONNX模型。

最后，我们使用Python编写的自定义推理代码将ONNX模型与Azure Functions一起使用Azure CLI部署。ONNX支持大多数常用的ML框架和工具。需要注意的一点是，如果您希望在不同的目标部署框架上使用不同的ML模型，例如从PyTorch转换为TensorFlow，将ML模型转换为ONNX格式是很有用的。如果您在源和目标上使用相同的框架，那么您不需要将模型转换为ONNX格式。

以下图表说明了这种方法的架构。

我们使用SageMaker Studio笔记本和SageMaker Python SDK来构建和训练我们的模型。SageMaker Python SDK是一个用于在SageMaker上训练和部署ML模型的开源库。有关更多详细信息，请参阅创建或打开Amazon SageMaker Studio笔记本。

以下部分中的代码片段在SageMaker Studio笔记本环境中使用Data Science 3.0镜像和Python 3.0内核进行了测试。

在此解决方案中，我们演示了以下步骤：

1. 训练PyTorch模型。
2. 将PyTorch模型导出为ONNX模型。
3. 打包模型和推理代码。
4. 将模型部署到Azure Functions。

先决条件

您应具备以下先决条件：

• 一个AWS账户。
• 一个SageMaker域和SageMaker Studio用户。有关创建这些的说明，请参阅使用快速设置加入Amazon SageMaker域。
• Azure CLI。
• 访问Azure的权限以及具有创建和管理Azure Functions权限的服务主体的凭据。

使用PyTorch训练模型

在本节中，我们详细介绍训练PyTorch模型的步骤。

安装依赖项

安装库以执行模型训练和模型部署所需的步骤：

pip install torchvision onnx onnxruntime

完成初始设置

我们首先导入Python的AWS SDK（Boto3）和SageMaker Python SDK。作为设置的一部分，我们定义了以下内容：

• 一个会话对象，提供了在SageMaker和我们自己的帐户上下文中使用的便利方法。
• 用于委派训练和托管服务权限的SageMaker角色ARN。我们需要这个角色，以便这些服务可以访问存储我们的数据和模型的S3存储桶。有关创建满足您业务需求的角色的说明，请参阅SageMaker角色。对于本文，我们使用与Studio笔记本实例相同的执行角色。我们通过调用sagemaker.get_execution_role()来获取此角色。
• 我们的训练作业将运行的默认区域。
• 我们用于存储模型输出的默认存储桶和前缀。

请参阅以下代码：

import sagemaker
import boto3
import os

execution_role = sagemaker.get_execution_role()
region = boto3.Session().region_name
session = sagemaker.Session()
bucket = session.default_bucket()
prefix = "sagemaker/mnist-pytorch"

创建训练数据集

我们使用公共存储桶sagemaker-example-files-prod-{region}中可用的数据集。数据集包含以下文件：

• train-images-idx3-ubyte.gz – 包含训练集图像
• train-labels-idx1-ubyte.gz – 包含训练集标签
• t10k-images-idx3-ubyte.gz – 包含测试集图像
• t10k-labels-idx1-ubyte.gz – 包含测试集标签

我们使用torchvision.datasets模块从公共存储桶下载数据，并在上传到我们的训练数据存储桶之前将其保存在本地。我们将此存储桶位置作为输入传递给SageMaker训练作业。我们的训练脚本使用此位置下载和准备训练数据，然后训练模型。请参阅以下代码：

MNIST.mirrors = [
    f"https://sagemaker-example-files-prod-{region}.s3.amazonaws.com/datasets/image/MNIST/"
]

MNIST(
    "data",
    download=True,
    transform=transforms.Compose(
        [transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))]
    ),
)

创建训练脚本

使用SageMaker，您可以使用脚本模式引入自己的模型。脚本模式下，您可以使用预构建的SageMaker容器，并提供自己的训练脚本，其中包含模型定义以及任何自定义库和依赖项。SageMaker Python SDK将我们的脚本作为entry_point传递给容器，容器加载并运行提供的脚本中的训练函数来训练我们的模型。

训练完成后，SageMaker将模型输出保存在我们提供给训练作业的S3存储桶中。

我们的训练代码改编自以下PyTorch示例脚本。以下代码摘录显示了模型定义和训练函数：

# 定义网络

class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.conv1 = nn.Conv2d(1, 32, 3, 1)
        self.conv2 = nn.Conv2d(32, 64, 3, 1)
        self.dropout1 = nn.Dropout(0.25)
        self.dropout2 = nn.Dropout(0.5)
        self.fc1 = nn.Linear(9216, 128)
        self.fc2 = nn.Linear(128, 10)

    def forward(self, x):
        x = self.conv1(x)
        x = F.relu(x)
        x = self.conv2(x)
        x = F.relu(x)
        x = F.max_pool2d(x, 2)
        x = self.dropout1(x)
        x = torch.flatten(x, 1)
        x = self.fc1(x)
        x = F.relu(x)
        x = self.dropout2(x)
        x = self.fc2(x)
        output = F.log_softmax(x, dim=1)
        return output

# 训练

def train(args, model, device, train_loader, optimizer, epoch):
    model.train()
    for batch_idx, (data, target) in enumerate(train_loader):
        data, target = data.to(device), target.to(device)
        optimizer.zero_grad()
        output = model(data)
        loss = F.nll_loss(output, target)
        loss.backward()
        optimizer.step()
        if batch_idx % args.log_interval == 0:
            print('Train Epoch: {} [{}/{} ({:.0f}%)]\tLoss: {:.6f}'.format(
                epoch, batch_idx * len(data), len(train_loader.dataset),
                100. * batch_idx / len(train_loader), loss.item()))
            if args.dry_run:
                break

训练模型

现在，我们已经设置好了环境并创建了输入数据集和自定义训练脚本，我们可以使用SageMaker开始模型训练。我们使用SageMaker Python SDK中的PyTorch estimator在SageMaker上启动一个训练作业。我们将必需的参数传递给estimator并调用fit方法。当我们在PyTorch estimator上调用fit时，SageMaker使用我们的脚本作为训练代码开始训练作业：

from sagemaker.pytorch import PyTorch

output_location = f"s3://{bucket}/{prefix}/output"
print(f"训练结果将上传到：{output_location}")

hyperparameters={
    "batch-size": 100,
    "epochs": 1,
    "lr": 0.1,
    "gamma": 0.9,
    "log-interval": 100
}

instance_type = "ml.c4.xlarge"
estimator = PyTorch(
    entry_point="train.py",
    source_dir="code",  # 您的训练脚本所在的目录
    role=execution_role,
    framework_version="1.13",
    py_version="py39",
    instance_type=instance_type,
    instance_count=1,
    volume_size=250,
    output_path=output_location,
    hyperparameters=hyperparameters
)

estimator.fit(inputs={
    'training': f"{inputs}",
    'testing':  f"{inputs}"
})

导出已训练的模型为ONNX模型

训练完成并且我们的模型已保存到Amazon S3中预先定义的位置后，我们使用ONNX运行时将模型导出为ONNX模型。

我们在训练脚本中包含将模型导出为ONNX的代码，在训练完成后运行该代码。

PyTorch通过使用我们的输入运行模型并记录用于计算输出的操作的跟踪来将模型导出为ONNX。我们使用正确类型的随机输入和PyTorch的torch.onnx.export函数将模型导出为ONNX。我们还指定了输入的第一个维度为动态，以便在推断期间，我们的模型接受可变的batch_size输入。

def export_to_onnx(model, model_dir, device):
    logger.info("将模型导出为ONNX格式。")
    dummy_input = torch.randn(1, 1, 28, 28).to(device)
    input_names = [ "input_0" ]
    output_names = [ "output_0" ]
    path = os.path.join(model_dir, 'mnist-pytorch.onnx')
    torch.onnx.export(model, dummy_input, path, verbose=True, input_names=input_names, output_names=output_names,
                     dynamic_axes={'input_0' : {0 : 'batch_size'},    # 可变长度的轴
                                'output_0' : {0 : 'batch_size'}})

ONNX是一个用于深度学习模型的开放标准格式，可以实现PyTorch、Microsoft Cognitive Toolkit (CNTK)等深度学习框架之间的互操作性。这意味着您可以使用这些框架之一来训练模型，并随后将训练好的模型导出为ONNX格式。通过将模型导出为ONNX格式，您可以在更广泛的部署设备和平台上使用模型。

下载并提取模型文件

我们的训练脚本保存的ONNX模型已经被SageMaker复制到了我们在启动训练作业时指定的输出位置的Amazon S3上。模型文件以一个名为model.tar.gz的压缩归档文件的形式存储。我们将这个归档文件下载到本地目录中，并提取出其中的模型文件，即ONNX模型。

import tarfile

local_model_file = 'model.tar.gz'
model_bucket,model_key = estimator.model_data.split('/',2)[-1].split('/',1)
s3 = boto3.client("s3")
s3.download_file(model_bucket,model_key,local_model_file)

model_tar = tarfile.open(local_model_file)
model_file_name = model_tar.next().name
model_tar.extractall('.')
model_tar.close()

验证ONNX模型

我们的训练脚本将ONNX模型导出到一个名为mnist-pytorch.onnx的文件中。在下载并提取了这个文件之后，我们可以选择使用onnx.checker模块对ONNX模型进行验证。该模块中的check_model函数用于检查模型的一致性，如果测试失败，则会引发异常。

import onnx

onnx_model = onnx.load("mnist-pytorch.onnx")
onnx.checker.check_model(onnx_model)

打包模型和推理代码

在本文中，我们使用.zip部署方式用于Azure Functions。在这种方法中，我们将模型、相关代码和Azure Functions的设置打包到一个.zip文件中，并将其发布到Azure Functions。以下代码展示了我们部署包的目录结构：

mnist-onnx ├── function_app.py ├── model │ └── mnist-pytorch.onnx └── requirements.txt

列出依赖项

我们在部署包的根目录中的requirements.txt文件中列出了推理代码的依赖项。在发布包时，该文件用于构建Azure Functions的运行环境。

azure-functions numpy onnxruntime

编写推理代码

我们使用Python编写以下推理代码，使用ONNX Runtime库加载我们的模型并进行推理。这指示Azure Functions应用程序将端点可用于相对路径/classify。

import logging
import azure.functions as func
import numpy as np
import os
import onnxruntime as ort
import json


app = func.FunctionApp()

def preprocess(input_data_json):
    # 将JSON数据转换为输入张量
    return np.array(input_data_json['data']).astype('float32')
    
def run_model(model_path, req_body):
    session = ort.InferenceSession(model_path)
    input_data = preprocess(req_body)
    logging.info(f"输入数据的形状为 {input_data.shape}。")
    input_name = session.get_inputs()[0].name  # 获取模型的第一个输入的ID
    try:
        result = session.run([], {input_name: input_data})
    except (RuntimeError) as e:
        print("形状={0}，错误={1}".format(input_data.shape, e))
    return result[0] 

def get_model_path():
    d=os.path.dirname(os.path.abspath(__file__))
    return os.path.join(d , './model/mnist-pytorch.onnx')

@app.function_name(name="mnist_classify")
@app.route(route="classify", auth_level=func.AuthLevel.ANONYMOUS)
def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP 触发器函数正在处理请求。')
    # 从POST请求中获取img值
    try:
        req_body = req.get_json()
    except ValueError:
        pass

    if req_body:
        # 运行模型
        result = run_model(get_model_path(), req_body)
        # 将输出映射为整数并返回结果字符串
        digits = np.argmax(result, axis=1)
        logging.info(type(digits))
        return func.HttpResponse(json.dumps({"digits": np.array(digits).tolist()}))
    else:
        return func.HttpResponse(
             "这个HTTP触发器函数已成功执行。",
             status_code=200
        )

将模型部署到Azure Functions

现在我们已经将代码打包成所需的.zip格式，准备好将其发布到Azure Functions。我们使用Azure CLI来完成这个操作，Azure CLI是一个命令行实用程序，用于创建和管理Azure资源。使用以下代码安装Azure CLI：

!pip install -q azure-cli

然后按照以下步骤完成操作：

1. 登录Azure：

   !az login

2. 设置资源创建参数：

   import random
   
   random_suffix = str(random.randint(10000,99999))
   resource_group_name = f"multicloud-{random_suffix}-rg"
   storage_account_name = f"multicloud{random_suffix}"
   location = "ukwest"
   sku_storage = "Standard_LRS"
   functions_version = "4"
   python_version = "3.9"
   function_app = f"multicloud-mnist-{random_suffix}"

3. 使用以下命令创建Azure Functions应用以及所需的前提资源：

   !az group create --name {resource_group_name} --location {location}
   !az storage account create --name {storage_account_name} --resource-group {resource_group_name} --location {location} --sku {sku_storage}
   !az functionapp create --name {function_app} --resource-group {resource_group_name} --storage-account {storage_account_name} --consumption-plan-location "{location}" --os-type Linux --runtime python --runtime-version {python_version} --functions-version {functions_version}

4. 设置Azure Functions，以便在部署函数包时使用requirements.txt文件构建应用程序依赖项：

   !az functionapp config appsettings set --name {function_app} --resource-group {resource_group_name} --settings @./functionapp/settings.json

5. 配置函数应用程序以运行Python v2模型，并在.zip部署后对其接收的代码进行构建：

   {
       "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
       "SCM_DO_BUILD_DURING_DEPLOYMENT": true
   }

6. 在资源组、存储容器和具有正确配置的函数应用程序准备好后，将代码发布到函数应用程序：

   !az functionapp deployment source config-zip -g {resource_group_name} -n {function_app} --src {function_archive} --build-remote true

测试模型

我们已将ML模型部署到Azure Functions作为HTTP触发器，这意味着我们可以使用函数应用程序的URL向函数发送HTTP请求以调用函数并运行模型。

为了准备输入，从SageMaker示例文件桶中下载测试图像文件，并按照模型要求的格式准备一组样本：

from torch.utils.data import DataLoader
from torchvision import datasets, transforms
import matplotlib.pyplot as plt

transform=transforms.Compose(
        [transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))]
)

test_dataset = datasets.MNIST(root='../data',  download=True, train=False, transform=transform)
test_loader = DataLoader(test_dataset, batch_size=16, shuffle=True)

test_features, test_labels = next(iter(test_loader))

使用requests库向推理端点发送POST请求以进行样本输入。推理端点的格式如下所示：

import requests, json

def to_numpy(tensor):
    return tensor.detach().cpu().numpy() if tensor.requires_grad else tensor.cpu().numpy()

url = f"https://{function_app}.azurewebsites.net/api/classify"
response = requests.post(url, 
                json.dumps({"data":to_numpy(test_features).tolist()})
            )
predictions = json.loads(response.text)['digits']

清理

完成模型测试后，删除资源组以及包含的资源，包括存储容器和函数应用程序：

!az group delete --name {resource_group_name} --yes

此外，建议在SageMaker Studio中关闭空闲资源以减少成本。有关更多信息，请参阅通过自动关闭Amazon SageMaker Studio中的空闲资源来节省成本。

结论

在这篇文章中，我们展示了如何使用SageMaker构建和训练机器学习模型，并将其部署到另一个云提供商。在解决方案中，我们使用了SageMaker Studio笔记本，但对于生产工作负载，我们建议使用MLOps创建可重复训练工作流程，以加速模型开发和部署。

本文没有展示在多云环境中部署和运行模型的所有可能方式。例如，您还可以将模型打包到容器镜像中，连同推理代码和依赖库，以在任何平台上将模型作为容器化应用程序运行。有关此方法的更多信息，请参阅使用Amazon CodeCatalyst在多云环境中部署容器应用程序。本文的目的是展示组织如何在多云环境中使用AWS AI/ML功能。