LotteON 如何利用 Amazon SageMaker 和 MLOps 构建个性化推荐系统

_作者：SeungBum Shim、HyeKyung Yang、Jesam Kim、Jieun Lim 和 Gonsoo Moon，发表于 2024 年 5月 16 日, 在 和机器学习运维 (MLOps) 改善其推荐服务。

问题定义

传统的推荐服务主要通过识别产品之间的关系来提供与客户所选产品高度相关的产品。然而，我们意识到需要升级推荐服务，以便深入分析每位客户的喜好和需求。因此，我们决定引入基于深度学习的推荐算法，这种算法不仅能识别数据中的线性关系，还能捕捉更复杂的关系。为此，我们构建了 MLOps 架构来管理模型，并提供实时服务。

另一个需求是建立一个能够与现有推荐平台使用的代码库 GitLab 集成的持续集成和持续交付 (CI/CD) 管道，以便添加新开发的推荐模型，并创建一个可以通过定期再训练和重新分发模型持续改进服务质量的结构。

接下来的部分将介绍我们构建的 MLOps 平台，以便为客户提供高质量的推荐服务，以及实时推断基于深度学习的推荐算法（）的整体过程，并将其引入 LotteON。

解决方案架构

下图展示了作为 MLOps 的 Neural Collaborative Filtering （NCF）算法推荐模型的解决方案架构。主要使用的 AWS服务有 SageMaker、、、 (Amazon S3)、、 和 。我们结合多种 AWS 服务使用 ，并围绕以下组件设计架构：

删除)

代码准备和数据预处理

准备和预处理阶段包括以下步骤：

1. 数据科学家将包含模型和训练流水线的部署代码发布到 ，而 则将代码上传至 Amazon S3。
2. EMR 预处理批次通过 根据指定时间表运行。预处理的数据被加载到 MongoDB 中，作为特征存储，同时与 Amazon S3 联合使用。

训练流水线和模型部署

模型训练和部署阶段包括以下步骤：

1. 在数据上传至 Amazon S3 后，CodeBuild 根据 EventBridge 中指定的规则运行。
2. SageMaker 为 CodeBuild 预定义的流水线依次运行步骤，例如预处理、模型训练和模型注册。
3. 当训练完成后（通过 Lambda 步骤），已部署的模型将更新至 SageMaker 端点。

实时推荐推断

推断阶段包括以下步骤：

1. 客户端应用向 API 网关发起推断请求。
2. API 网关将请求发送至 Lambda，Lambda 再向 SageMaker 端点的模型发起推断请求以获取推荐列表。
3. Lambda 接收到推荐列表并将其提供给 API 网关。
4. API 网关通过 Recommendation API 将推荐列表提供给客户端应用。

使用 NCF 的推荐模型

NCF 是一款基于 2017年国际万维网会议上发布的论文的算法。该算法以基于神经网络的协同过滤为基础，克服了线性矩阵分解在传统推荐系统中的局限性。通过神经网络加入非线性因素，作者能够建模用户与商品之间更复杂的关系。NCF的数据来源于用户与商品的交互数据，整体模型结构如下

删除)

尽管 NCF 的模型架构较为简单，但其表现良好，因此我们选择它作为 MLOps 平台的原型。关于该模型的更多信息，请参考论文 。

接下来的部分将讨论此解决方案如何帮助我们构建上述的 MLOps 组件：

• 数据预处理
• 自动化模型训练和部署
• 实时推断通过模型服务
• CI/CD 结构

MLOps 组件 1：数据预处理

对于 NCF，我们使用用户与商品的交互数据，这需要大量资源来处理从应用收集的原始数据并将其转换为适合学习的格式。借助提供 Apache Hadoop 和 Spark 等完全托管环境的 Amazon EMR，我们能够加快数据处理。

数据预处理批次是通过编写 shell 脚本，使用 （AWSCLI）命令运行 Amazon EMR，然后将其注册到 Airflow，以便在特定的时间间隔运行。当预处理批次完成后，所需的训练/测试数据将根据运行时间进行分区并存储在 Amazon S3 中。以下是运行 Amazon EMR 的 AWS CLI 命令示例：

`bash aws emr create-cluster --release-label emr-6.0.0 \ --name "CLUSTER_NAME" \ --applications Name=Hadoop Name=Hive Name=Spark \ --tags 'Name=EMR-DATA- PREP' 'Owner=MODEL' 'Service=LOTTEON' \ --ec2-attributes '{"KeyName":"keyname","InstanceProfile":"DefaultRole","ServiceAccessSecurityGroup":"sg- xxxxxxxxxxxxxx","SubnetId":"subnet- xxxxxxxxxxxxxx","EmrManagedSlaveSecurityGroup":"sg- xxxxxxxxxxxxxx","EmrManagedMasterSecurityGroup":"sg-xxxxxxxxxxxxxx"}'
--instance-groups ' 文件，该文件在 CodeBuild运行时不可或缺。这些修改包括需要训练的数据的路径、版本和分区信息，都是成功运行训练流水线的关键。EventBridge 的后续角色是派发事件，该事件由 buildspec.yml 文件的更改引发，导致运行 CodeBuild。

CodeBuild 负责构建源代码，其定义了 SageMaker 流水线。在整个过程中，它参考 buildspec.yml文件，并运行克隆源代码、安装构建所需库等过程。我们可以通过 CodeBuild 控制台的 Project Build 选项卡查看构建的成功与失败历史，以及 SageMaker 流水线当前的实时日志。

SageMaker 流水线训练

SageMaker Pipelines 帮助您定义 ML 服务所需的 ，例如预处理、训练和部署。在 SageMaker Studio中，您可以可视化每个步骤，这对于管理模型非常有用，您还可以管理经过训练的模型和端点的历史记录。您可以通过将条件语句附加到步骤结果来设置步骤，以便仅采用具有良好再训练结果的模型或为学习失败做准备。我们的流水线包含以下高层步骤：

• 模型训练
• 模型注册
• 模型创建
• 模型部署

每个步骤都在 中可视化，您还可以实时查看每一步的结果或进展，如下图所示。

删除)

让我们通过一些代码示例从模型训练到部署逐步介绍。

训练模型

首先，您需要定义一个 以用于训练，并设置训练步骤。这需要您提前准备好训练代码（例如，train.py），并将代码位置作为 source_dir 的参数传递。训练步骤运行您作为 entry_point 参数传递的训练代码。默认为实例中指定的容器启动训练，您需要提供所开发的训练 Docker镜像的路径。但是，如果您在这里指定了估算器的框架，您可以传入所用框架和 Python 版本的版本号，它将自动从 Amazon ECR获取相应版本的容器镜像。

定义完 PyTorch Estimator 后，您需要定义训练所涉及的步骤。您可以通过将之前定义的 PyTorch Estimator作为参数传递，并设置输入数据的位置来实现。当您传入输入数据的位置时，SageMaker 训练作业会将训练和测试数据下载至容器中的特定路径，格式为 /opt/ml/input/data/<channel_name>（例如，/opt/ml/input/data/train）。

此外，在定义 PyTorch Estimator 时，可以使用 来监控在模型训练时生成的学习指标，这些指标可通过 监控。您还可以通过指定 estimator_output_path 来指定模型训练结果存储位置，并通过 model_hyperparameters 指定模型训练所需的参数。以下是代码示例：

metric_definitions=[ {'Name': 'HR', 'Regex': 'HR=(._?);'}, {'Name': 'NDCG',
'Regex': 'NDCG=(._?);'}, {'Name': 'Loss', 'Regex': 'Loss=(.*?);'} ]
estimator_output_path = f's3://{bucket}/{prefix}' model_hyperparameter =
{'epochs': 10, 'lr': 0.001, 'batch_size': 256, 'top_k' : 10, 'dropout' : 0.3,
'factor_num' : 32, 'num_layers' : 3 }  
s3_code_uri = 's3://code_location/source.tar.gz'

host_estimator = PyTorch( entry_point="train.py", source_dir = s3_code_uri,
output_path = estimator_output_path, role=aws_role, framework_version='1.8.1',
py_version='py3', instance_count=1, instance_type='ml.p3.2xlarge', session =
pipeline_session, hyperparameters=model_hyperparameter, metric_definitions =
metric_definitions )

from sagemaker.inputs import TrainingInput from sagemaker.workflow.stepsimport TrainingStep

data_loc = f's3://{bucket}/{prefix}'

step_train = TrainingStep( name= "NCF-Training", estimator=host_estimator,
inputs={ "train": TrainingInput(s3_data=data_loc), "test":
TrainingInput(s3_data=data_loc),  
} ) ```

### 创建模型包组

下一步是创建模型包组以管理训练好的模型。通过在模型包中注册训练好的模型，您可以按版本进行管理，如下图所示。这一过程只需在首次训练模型时进行一次，后续可以继续添加和更新模型，只需声明相同的组名即可。

删除)

以下是相关代码：

```python import boto3

model_package_group_name = 'NCF' sm_client = boto3.client("sagemaker")
model_package_group_input_dict = { "ModelPackageGroupName" :
model_package_group_name, "ModelPackageGroupDescription" : "Model PackageGroup" }

response =
sm_client.list_model_package_groups(NameContains=model_package_group_name) iflen(response['ModelPackageGroupSummaryList']) == 0:
create_model_package_group_response =
sm_client.create_model_package_group(**model_package_group_input_dict) ```

### 将训练好的模型添加到模型包组

接下来，将训练好的模型添加到您创建的模型包组。在下面的代码中，当您声明 Model类时，您将获得上一步模型训练的结果，从而创建步骤间的依赖关系。具有声明依赖关系的步骤只有在前一步成功时才能运行。不过，您可以使用
 选项，声明步骤之间的依赖关系，即使数据无因果关系。

当训练好的模型在模型包组中注册后，您可以使用这些信息管理和跟踪未来的模型版本，创建一个 、执行

等。

```python from sagemaker.workflow.model_step import ModelStep fromsagemaker.model import Model

inference_image_uri = '763104351884.dkr