在当今的数据驱动世界中，机器学习模型对于做出实时决策、优化流程以及获得有价值的见解至关重要。然而，开发模型仅是整个流程的一部分。在生产环境中管理机器学习模型需要具备可扩展性、监控和自动化，以确保模型能持续提供价值。这就到了MLOps（机器学习运维）大显身手的时候了。
MLOps,通过实现模型的无缝集成、持续部署和监控，弥合了机器学习开发与实际应用之间的差距。借助 AWS 提供的全套工具，实现一个稳健、可扩展且自动化的 MLOps 流程变得比以往任何时候都更加容易。
在这篇博客里，我们将探讨如何使用 AWS 服务（如 SageMaker、Lambda、Step Functions 和 CodePipeline）搭建整个 MLOps 管道。我们将通过一个零售需求预测案例来进行说明，提供详细的代码示例，并概述运行模型的最佳实践，无论是本地还是生产环境。
使用MLOps进行零售需求预测的业务场景（MLOps，即机器学习运维）想象你在一家零售连锁公司工作，需要预测各个分店的产品需求。准确地预测不同店铺的产品需求可以帮助优化库存管理，减少浪费，避免断货。MLOps 管道需要处理以下内容：
- 数据摄取：收集销售数据、天气数据、节假日信息和促销活动信息。
- 数据预处理：清理、转换和标准化数据。
- 模型训练：训练预测用的机器学习模型。
- 模型部署：在生产环境中部署模型，以供实时API预测使用。
- 监控与再训练：监控模型性能，并在检测到漂移时触发再训练。
为了简化这个过程，我们将设置MLOps流水线在AWS上，并将其部分流程本地运行以进行开发和测试工作。
注：在中文里，“AWS”通常保持原样，不翻译。因此，原文中的“AWS”在翻译中也保留为“AWS”。
详细项目结构:为了构建一个可用于生产的MLOps流水线，让我们先定义项目的结构。
mlops-forecasting
│
├── 数据/ # 存储原始和处理过的数据
│ ├── 原始/ # 原始销售数据
│ └── 处理过的/ # 处理和清洗过的数据
│
├── 模型目录/ # 保存模型的目录
│ └── model.joblib # 训练好的模型
│
├── 笔记本/ # 用于实验的 Jupyter 笔记本
│ └── exploratory.ipynb
│
├── 源码目录/ # 机器学习管道的源码
│ ├── preprocessing.py # 数据清洗和转换
│ ├── train.py # 训练脚本
│ ├── inference.py # 推理逻辑
│ └── monitor.py # 监控和漂移检测
│
├── 脚本目录/ # CI/CD 和自动化使用的脚本
│ ├── buildspec.yml # AWS CodeBuild 配置
│ └── deploy.sh # 部署脚本文件
│
├── 配置目录/ # 环境配置文件
│ └── config.yml # 超参数和环境设置
│
├── 测试目录/ # 管道的单元测试
│ ├── test_preprocessing.py
│ ├── test_train.py
│ └── test_inference.py
│
├── terraform/ # AWS 设置的基础设施即代码
│ ├── s3.tf # S3 存储桶设置
│ ├── sagemaker.tf # SageMaker 模型部署
│ └── monitoring.tf # 模型监控设置
│
├── Dockerfile # 用于容器化开发的 Dockerfile
├── requirements.txt # Python 依赖项列表
└── README.md # 项目文档一步一步的工作流程
1. 数据导入和预处理 (位于src/preprocessing.py文件中)我们从清理和转换销售数据开始。
import pandas as pd
def preprocess_data(input_path, output_path):
# 加载原始销售数据
df = pd.read_csv(input_path)
# 删除缺失的数据
df.dropna(inplace=True)
# 特征处理：销售数据的移动平均
df['sales_moving_avg'] = df['sales'].rolling(window=7).mean()
# 标准化销售数据
df['sales'] = (df['sales'] - df['sales'].mean()) / df['sales'].std()
# 保存处理过的数据
df.to_csv(output_path, index=False)
if __name__ == "__main__":
# 主函数入口
preprocess_data('data/raw/sales_data.csv', 'data/processed/cleaned_sales_data.csv')你可以在本地运行这个脚本试试看。
在命令行中运行以下命令：python src/preprocessing.py
最佳实践: 确保进行数据验证，避免这些问题影响你的数据处理流程。
2. 模型训练和版本控制 (src/train.py)现在，让我们用随机森林模型来预测销售额。
从sklearn.ensemble导入RandomForestRegressor as rfr
从sklearn.model_selection导入train_test_split
从sklearn.metrics导入mean_absolute_error
导入joblib
导入pandas as pd
def train_model(data_path, model_path):
df = pd.read_csv(data_path)
# 将数据拆分为特征和目标变量
X = df.drop(columns=['sales'])
y = df['sales']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
# 训练随机森林回归模型
rfr = RandomForestRegressor(n_estimators=100)
rfr.fit(X_train, y_train)
# 评估模型
y_pred = rfr.predict(X_test)
mae = mean_absolute_error(y_test, y_pred)
print(f"模型的MAE: {mae}")
# 保存训练后的模型
joblib.dump(rfr, model_path)
如果 __name__ == '__main__':
train_model('data/processed/cleaned_sales_data.csv', 'models/model.joblib')运行训练脚本吧。
运行Python脚本的命令是 "python src/train.py"，这将启动训练脚本。
最佳做法: 始终进行 交叉验证 和 超参数调整 来优化模型的表现。
3. 单元测试（简称单元测）在MLOps中，测试至关重要，以确保每个组件的完整性不受损害。这里有一个预处理测试的例子。
import pandas as pd
from src.preprocessing import preprocess_data
def test_preprocessing():
# 初始化数据
data = {'sales': [200, 300, 400, None, 500]}
df = pd.DataFrame(data)
# 对数据进行预处理并保存到文件
preprocess_data(df, 'data/processed/test_cleaned_data.csv')
processed_df = pd.read_csv('data/processed/test_cleaned_data.csv')
# 确保预处理后的数据中没有缺失值
assert processed_df.isnull().sum().sum() == 0使用 pytest 运行测试时
pytest tests/test_preprocessing.py最佳做法: 确保所有模块（包括预处理、训练和推理）都有相应的单元测试，并保持较高的测试覆盖率。
4. Docker 用于本地开发为了确保本地开发环境的一致性，请利用 Docker 将项目打包成容器。
Dockerfile 文件:
FROM python:3.8-slim
WORKDIR /app
COPY . /app
RUN pip install -r requirements.txt
CMD ["python", "src/train.py"]构建 Docker 镜像：
运行以下Docker构建命令，将镜像标记为 mlops-forecasting：
docker build -t mlops-forecasting .请运行容器。
运行容器(Note: The code segments are kept as in the original, but the fluent and polite version of the translated sentence is placed before the code block.)
docker run -v $(pwd)/data:/app/data mlops-forecasting运行Docker容器，并将当前目录下的data文件夹挂载到容器内的/app/data路径，最后运行mlops-forecasting镜像。
**最佳做法:** 使用 Docker 中的 **多阶段构建过程** 来来优化镜像的大小和安全性。
## 5. 使用 AWS CodePipeline 实现 CI/CD
自动化的训练和部署流程，可以设置**AWS CodePipeline**和**CodeBuild**。
**buildspec.yml:**
version: 0.2
phases:
安装阶段：
命令：
- pip install -r requirements.txt
构建阶段：
命令： - python src/train.py
后构建阶段：
命令： - aws s3 cp models/model.joblib s3://my-bucket/model/
artifacts:
输出文件：
文件： - models/model.joblib
在部署模型之前，确保CI/CD管道包含自动化的验证和测试。最佳做法:
6. 在SageMaker上的部署和监控将训练好的模型部署好以便进行实时预测，并实施SageMaker的漂移监控。
import sagemaker
from sagemaker.sklearn import SKLearnModel
model_uri = 's3://my-bucket/model/model.joblib'
model = SKLearnModel(model_data=model_uri,
role='arn:aws:iam::123456789012:role/SageMakerRole',
entry_point='src/inference.py')
predictor = model.deploy(instance_type='ml.m5.large', initial_instance_count=1)要使用SageMaker 模型监控设置漂移检测功能，请按照以下步骤操作：
导入sagemaker.model_monitor中的DataCaptureConfig配置
data_capture_config = DataCaptureConfig(
enable_capture=True, # 开启捕获功能
sampling_percentage=100, # 采样百分比设置为100表示捕获所有数据
destination_s3_uri='s3://my-bucket/data-capture/' # 目标S3 URI
)
预测器.update_data_capture_config(data_capture_config=data_capture_config)最佳做法: 启用模型可解释能力和偏差检测，从而确保预测结果的公平性。
总结在这篇博客中，我们讨论了如何在AWS上从头到尾构建一个全面的MLOps流水线，涵盖从本地开发到部署和监控的全过程。通过遵循业界公认的优秀实践，如模块化、版本控制、自动化监控和安全地部署，你的MLOps流水线将变得可扩展、自动化，并且准备好投入生产。从小规模开始，先建立本地环境，之后逐步将MLOps流水线迁移到AWS，以实现机器学习模型的全生命周期自动化管理。MLOps（机器学习操作）。
需要帮忙解决 MLOps 的问题吗？有问题可以联系 ThamesTech AI 咨询一下。
如果您正在考虑实施MLOps流水线，或在AWS SageMaker，机器学习或云基础设施方面需要咨询服务，ThamesTech AI可以提供帮助。我们的团队专长于创建符合您业务需求的可扩展，自动化和生产就绪的机器学习解决方案。
请联系我们咨询一下
- 网站: thamestech.ai
- 电子邮件: info@thamestech.ai
让我们帮助您加速您通往AI驱动成功的旅程!
[OOP]: 面向对象编程
[CRUD]: CRUD
[JVM]: Java虚拟机
[SUT]: 测试系统