MLOPS

维基百科：MLOps是ModelOps的子集，是数据科学家和操作专业人员之间进行协作和交流的一种做法，可帮助管理生产机器学习生命周期。与DevOps或DataOps方法相似，MLOps旨在提高自动化程度并提高生产ML的质量，同时还关注业务和法规要求。

MLOps的概念本身也在发展之中，不同的公司对他的描述也不一样,但基本都是沿用DevOps的思路来定义。

Google：MLOps是一种机器学习工程文化和做法，旨在统一机器学习系统开发（Dev）和机器学习系统运维（Ops）。实施MLOps意味着您*将在机器学习系统构建流程的所有步骤（包括集成、测试、发布、部署和基础架构管理）中实现自动化和监控。

Microsoft：MLOps是基于可提高工作流效率的DevOps原理和做法。目标是更快试验和开发模型，更快地将模型部署到生产环境，质量保证和端到端世系跟踪。

Amazon：ML考虑了AI/ML项目在项目管理、CI/CD和质量保证方面的独特方面，帮助客户缩短交付时间，减少缺陷并提高数据科学家的工作效率。

a4e2c59d03fd24cd41cf2a76f24f70ef.png

机器学习、数据工程和 DevOps 都在这个领域融合在一起。换句话说，它将机器学习与设计，构建和维护系统的任务联系起来。

fd847413e385b0a4c1caf8dd3e1321bc.png

MLflow 是什么？

MLflow 是一个开源平台，旨在管理整个机器学习生命周期。这包括实验跟踪、模型开发和选择、部署以及监控等方面。在 MLOps 中，MLflow 对于协作和自动化 ML 流程特别有价值。

MLflow VS Kubeflow

MLflow 和 Kubeflow 都是开源工具,旨在简化机器学习工作流程的管理,但它们的重点和使用场景有所不同。

1、MLflow 是一个用于管理机器学习生命周期的开源平台,包括实验跟踪、项目打包、模型管理等。 2、它侧重于为数据科学家提供一个组织代码、数据、模型以及运行结果的平台,并使之便于复制和共享。 3、MLflow 支持多种机器学习库,如 TensorFlow、PyTorch、XGBoost 等。 4、它由四个核心组件组成:Tracking、Projects、Models 和 Model Registry。

1、Kubeflow 是一个基于 Kubernetes 的开源机器学习平台,专注于机器学习工作流程的编排和部署。 2、它为机器学习工作流程提供了一个可移植、可扩展和部署友好的环境。 3、Kubeflow 提供了多个组件支持不同阶段的工作,如数据准备、训练、模型服务、notebook等。 4、它利用 Kubernetes 的优势,如容器编排、自动化扩展、故障恢复等,适用于生产环境部署。

相似之处:

1、两者都是用于支持和简化机器学习工作流程的开源工具。 2、它们都提供了跟踪实验、管理模型等功能。

不同之处:

1、MLflow 更侧重于为数据科学家提供一个组织和管理实验的平台,而 Kubeflow 更注重机器学习工作流的编排和生产部署。 2、MLflow 支持多种机器学习库,而 Kubeflow 基于 Kubernetes,为部署提供了更好的支持和可扩展性。 3、MLflow 通常用于较小规模的实验管理,而 Kubeflow 更适用于生产级别的大规模部署。

两者可以结合使用,MLflow 用于管理实验和模型,而 Kubeflow 则用于在 Kubernetes 上进行模型的部署和服务。它们为不同阶段的机器学习工作流程提供了支持和简化

使用场景

MLflow： 主要关注机器学习的生命周期管理，包括实验跟踪、模型管理、部署和监控。它适用于各种深度学习框架，提供自动日志记录、定制序列化以及统一的界面。MLflow 的模型注册表非常适合大型组织，其中不同团队可能在处理多种模型，提供版本控制、注释以及模型生命周期阶段的定义等特性。此外，MLflow 通过包括 Docker 和 GPU 支持的部署功能，确保了不同环境中的一致行为。

Kubeflow： 是一个全面的、开源的机器学习平台，旨在管理 Kubernetes 上的端到端机器学习工作流程。它提供了一个可伸缩且灵活的工具包，用于部署、监控和管理复杂的 ML 系统。Kubeflow 主要是为了在 Kubernetes 上编排整个 ML 工作流程，从数据预处理到模型训练、服务和监控。Kubeflow 通过其流水线提供了强大的解决方案，用于构建和部署可重复使用的 ML 工作流程，并支持连续集成和交付（CI/CD），便于快速实验和产品化。此外，Kubeflow 提供对多个 ML 框架的支持，例如 TensorFlow、PyTorch 和 XGBoost，并且其 Katib 组件自动化了超参数的调整。

总的来说，MLflow 在实验跟踪、模型管理和部署方面更加突出，特别适用于需要详细记录和比较实验结果的情况。而 Kubeflow 则更适用于需要在 Kubernetes 上进行端到端机器学习操作的复杂场景，特别是当涉及到分布式训练和可扩展的模型服务时

MLflow 的组件：

image.png

1. MLflow Tracking： 这是一个API和用户界面，可以在ML运行过程中记录参数、代码版本、指标和产品，并在以后可视化结果。它适应于任何环境，允许你记录到本地文件或服务器，并比较多个运行。团队还可以使用它来比较来自不同用户的结果。

2. MLflow Projects： 这是一种简化打包和重用数据科学代码的方式。每个项目是一个包含代码或Git仓库的目录，还有一个描述文件用于指定依赖关系和执行指令。当你使用Tracking API时，MLflow会自动跟踪项目的版本和参数，使得可以轻松地从GitHub或你的Git仓库运行项目，并将它们链接到多步骤的工作流中。

3. MLflow Models： 它允许你以不同的格式打包ML模型，并提供各种部署工具。每个模型保存为一个目录，其中包含一个描述文件列出其支持的格式。MLflow提供了将常见模型类型部署到各种平台的工具，包括基于Docker的REST服务器、Azure ML、AWS SageMaker和Apache Spark用于批处理和流媒体推理。当你使用Tracking API输出MLflow模型时，MLflow会自动跟踪其来源，包括项目和运行它们的来源。

4. MLflow Model Registry： 这是一个集中式模型存储，具有API和用户界面，用于协作管理MLflow模型的整个生命周期。它包括模型血统、版本控制、阶段转换和注释，以实现有效的模型管理。

5. MLflow recipes: 的前身是mlflow pipeline,mlflow pipeline 于 2022 年 11 月 7 日弃用,，用于机器学习中用到的各种步骤通过recipes编排形成一个可复用且容易拓展的mlflow套件。recipes可以缓存中间运行的结果，使得任务可以从失败节点开始执行，节省算力资源。

MLflow 的工作流程：

image.png

1. 实验： 通过记录参数、指标和输出来进行实验。MLflow UI 有助于跟踪和比较这些实验。
2. 打包： 使用 MLflow Projects 打包数据科学代码，实现一致的执行和协作。
3. 模型保存与服务： 使用 MLflow 统一格式保存模型，并通过多种机制（如本地 REST API 端点）提供服务。
4. 模型注册与部署： 在模型注册表中注册模型进行版本控制和部署。将模型部署到各种环境，如 Kubernetes、云平台或 Databricks。
5. 监控和维护： 部署后，使用 MLflow 监控模型性能并管理其生命周期。

如何使用 MLflow:

1. 设置： 安装 MLflow 并设置环境。这可能涉及配置跟踪服务器以进行协作工作。
2. 运行实验： 使用 MLflow 记录实验。这涉及跟踪每次运行的参数、指标和输出。
3. 模型开发和记录： 使用 MLflow 的 API 训练和记录模型，适用于各种 ML 框架。
4. 模型评估和可视化： 使用 SHAP 等工具评估模型，并使用 MLflow UI 可视化结果。
5. 部署： 使用 MLflow 的部署工具在多种环境中部署模型，如本地服务器或 Kubernetes。

在可视化方面，MLflow 在 UI 中支持以表格和图表形式显示指标历史，提供了对模型性能随时间的变化或在不同运行中的比较的清晰了解。

pip install mlflow
mlflow server
http://localhost:5000

docker启动：
docker pull ghcr.io/mlflow/mlflow:v2.7.1 
docker run -d -it --name mlflow_demo -p 5001:5000 -p 8081:8080 ghcr.io/mlflow/mlflow:v2.7.1 /bin/bash
docker cp mlflow-2.7.1 a4b3d1454fd1:/
docker exec -it a4b3d1454fd1 /bin/bash 

mkdir /mlflow_workspace
cd /mlflow_workspace/
## 运行模型训练实验, 使用本地python环境
export GIT_PYTHON_REFRESH=quiet ; mlflow run --env-manager local /mlflow-2.7.1/examples/sklearn_elasticnet_wine -P alpha=0.7

启动mlflow ui服务
nohup mlflow ui -h 0.0.0.0 -p 5000 > mlflow_ui.log 2>&1 &

启动模型服务
mlflow models serve --env-manager local -h 0.0.0.0   --port 8080   -m runs:/ee55b2f4bd844dc99bdff3865ffa3d43/model

测试

创建实验：

from mlflow import MlflowClient
#  mlflow.set_tracking_uri(uri="http://10.0.102.50:15000")   # 全局设置
client = MlflowClient(tracking_uri="http://10.0.102.50:15000") # 可与不同的跟踪服务器交互
all_experiments = client.search_experiments()

default_experiment = [
    {"name": experiment.name, "lifecycle_stage": experiment.lifecycle_stage}
    for experiment in all_experiments
    if experiment.name == "Default"
][0]
# Provide an Experiment description that will appear in the UI
experiment_description = (
    "This is the grocery forecasting project. "
    "This experiment contains the produce models for apples."
)

# Provide searchable tags that define characteristics of the Runs that
# will be in this Experiment
experiment_tags = {
    "project_name": "grocery-forecasting",
    "store_dept": "produce",
    "team": "stores-ml",
    "project_quarter": "Q3-2023",
    "mlflow.note.content": experiment_description,
}

# Create the Experiment, providing a unique name
produce_apples_experiment = client.create_experiment(
    name="Apple_Models", tags=experiment_tags
)

image.png

实验追踪

import torch
import torch.optim as optim
import torch.nn.functional as F
from torch.utils.data import DataLoader
from torchvision import datasets, transforms

import mlflow
import mlflow.pytorch

# 定义超参数
learning_rate = 0.01
batch_size = 128
epochs = 10

# 开始一个新的实验
mlflow.set_experiment("pytorch_mnist")

# 加载数据集
train_dataset = datasets.MNIST("data", train=True, download=True,
                               transform=transforms.Compose([
                                   transforms.ToTensor(),
                                   transforms.Normalize((0.1307,), (0.3081,))
                               ]))
test_dataset = datasets.MNIST("data", train=False, download=True,
                              transform=transforms.Compose([
                                  transforms.ToTensor(),
                                  transforms.Normalize((0.1307,), (0.3081,))
                              ]))
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=True)

# 定义模型
class Net(torch.nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.conv1 = torch.nn.Conv2d(1, 10, kernel_size=5)
        self.conv2 = torch.nn.Conv2d(10, 20, kernel_size=5)
        self.fc1 = torch.nn.Linear(320, 50)
        self.fc2 = torch.nn.Linear(50, 10)

    def forward(self, x):
        x = F.relu(F.max_pool2d(self.conv1(x), 2))
        x = F.relu(F.max_pool2d(self.conv2(x), 2))
        x = x.view(-1, 320)
        x = F.relu(self.fc1(x))
        x = self.fc2(x)
        return F.log_softmax(x, dim=1)

model = Net()

# 定义优化器和损失函数
optimizer = optim.SGD(model.parameters(), lr=learning_rate)
criterion = torch.nn.CrossEntropyLoss()

# 使用 MLflow 进行实验追踪
with mlflow.start_run():
    # 记录实验参数
    mlflow.log_param("learning_rate", learning_rate)
    mlflow.log_param("batch_size", batch_size)
    mlflow.log_param("epochs", epochs)

    for epoch in range(epochs):
        # 训练模型
        model.train()
        for batch_idx, (data, target) in enumerate(train_loader):
            optimizer.zero_grad()
            output = model(data)
            loss = criterion(output, target)
            loss.backward()
            optimizer.step()

        # 测试模型
        model.eval()
        test_loss = 0
        correct = 0
        with torch.no_grad():
            for data, target in test_loader:
                output = model(data)
                test_loss += criterion(output, target).item()
                pred = output.argmax(dim=1, keepdim=True)
                correct += pred.eq(target.view_as(pred)).sum().item()

        # 记录实验结果
        test_loss /= len(test_loader.dataset)
        test_accuracy = 100\. * correct / len(test_loader.dataset)
        # 调用mlflow api
        mlflow.log_metric("test_loss", test_loss, step=epoch)
        mlflow.log_metric("test_accuracy", test_accuracy, step=epoch)

        # 保存模型
        mlflow.pytorch.log_model(model, "models", epoch)

        # 输出实验结果
        print(f"Epoch {epoch+1}: test_loss={test_loss:.4f}, test_accuracy={test_accuracy:.2f}%")

使用 set_experiment 开始一个新的实验。使用 log_param 记录实验的超参数，使用 log_metric 记录实验结果，以及使用 log_model 记录模型。这些记录将被保存到 MLflow 服务器或本地文件系统中，以便查看和比较不同实验的结果。

在启动训练后，本地会出现一个mlruns 目录，它是 MLflow 默认的存储实验数据和结果的目录。MLflow 会将每个实验的信息保存到一个单独的子目录中，其中包括实验参数、指标、文件等等。mlruns 目录可以存储在本地文件系统中，也可以存储在远程服务器或云存储中。可以使用 MLflow UI 查看和比较不同实验的结果

使用 mlflow ui 命令可以启动 MLflow UI，以便查看和比较不同实验的结果。例如，假设将 mlruns 目录存储在本地文件系统的 /path/to/mlruns 目录中，我们可以使用以下命令启动 MLflow UI：

mlflow ui --backend-store-uri <mlruns文件夹所在的目录>

image.png

加载模型：

使用 mlflow.pytorch.load_model 函数加载之前保存的 PyTorch 模型。该函数接受三个参数：

1、model_uri：模型的 URI，可以是本地文件路径或远程服务器地址。 2、map_location：可选参数，指定模型应该加载到哪个设备上。 3、model：可选参数，指定模型的类型。如果未指定，则使用 torch.nn.Module。

import torch
import mlflow.pytorch

# 从 MLflow 加载模型
model = mlflow.pytorch.load_model("models")

# 如果有gpu的话
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
model.to(device)

# 使用模型进行推理
with torch.no_grad():
    inputs = torch.randn(1, 1, 28, 28).to(device)
    outputs = model(inputs)
    print(outputs)

加载模型时，需要指定模型的 URI，可以是本地文件路径或远程服务器地址。在前面的示例中，模型保存在 MLflow 中，其 URI 为 "models"。如果将模型保存在本地文件系统中，则可以将 URI 设置为文件路径，例如 "/path/to/pytorch_model"。

如果模型使用了 GPU 进行训练并保存在 GPU 上，需要使用 map_location 参数指定模型应该加载到哪个设备上。如上实例使用了 torch.device 函数指定了设备。

Projects

项目可以是远程存储库或本地目录。与MLflow模型不同，MLflow项目旨在实现机器学习项目的可移植性和分布性

MLflow项目由名为“MLProject”的一个YAML声明文件来定义，其中公开了相应项目的一系列规范内容。

模型实现的关键特征在MLProject文件中指定，这些特征包括：

• 模型接收的输入参数
• 参数的数据类型
• 用于执行所述模型的命令，以及
• 项目运行的环境

image.png

conda.yaml

name: tutorial
channels:
  - conda-forge
dependencies:
  - python=3.8
  - pip
  - pip:
      - scikit-learn==1.2.0
      - mlflow>=1.0
      - pandas

MLproject定义：

name: tutorial

python_env: python_env.yaml

entry_points:
  main:
    parameters:
      alpha: {type: float, default: 0.5}
      l1_ratio: {type: float, default: 0.1}
    command: "python train.py {alpha} {l1_ratio}"

mlflow run sklearn_elasticnet_wine -P alpha=0.5 // 本地
mlflow run git@github.com:FernandoLpz/MLflow-example.git -P tree_depth=3 // 远程

image.png

image.png

image.png

模型训练
$: mlflow run --env-manager local examples/sklearn_elasticnet_wine -P alpha=0.5

将模型打包成镜像
$: mlflow models build-docker --model-uri "runs:/f933482ecd664890a6fa097dfd6bdef7/model" --name "my-image-name" 

运行一个模型，指定服务端口
$: mlflow models serve --env-manager local -h 0.0.0.0 -m my_model

模型

MLflow模型允许将机器学习模型打包成标准格式，以便通过REST API、Microsoft Azure ML、Amazon SageMaker或Apache Spark等不同服务直接使用，打包方面，MLflow生成一个包含两个文件的目录，一个是模型，另一个是指定模型打包和加载细节的文件。如下。

image.png

artifact_path: model
flavors:
  python_function:
    env:
      conda: conda.yaml
      virtualenv: python_env.yaml
    loader_module: mlflow.sklearn
    model_path: model.pkl
    predict_fn: predict
    python_version: 3.8.18
  sklearn:
    code: null
    pickled_model: model.pkl
    serialization_format: cloudpickle
    sklearn_version: 1.2.0
mlflow_version: 2.11.1
model_size_bytes: 878
model_uuid: 49c54a35c1be4b53809427050e3463d5
run_id: 163415697d22464ba51c716d2eb68407
signature:
  inputs: '[{"type": "double", "name": "fixed acidity", "required": true}, {"type":
    "double", "name": "volatile acidity", "required": true}, {"type": "double", "name":
    "citric acid", "required": true}, {"type": "double", "name": "residual sugar",
    "required": true}, {"type": "double", "name": "chlorides", "required": true},
    {"type": "double", "name": "free sulfur dioxide", "required": true}, {"type":
    "double", "name": "total sulfur dioxide", "required": true}, {"type": "double",
    "name": "density", "required": true}, {"type": "double", "name": "pH", "required":
    true}, {"type": "double", "name": "sulphates", "required": true}, {"type": "double",
    "name": "alcohol", "required": true}]'
  outputs: '[{"type": "tensor", "tensor-spec": {"dtype": "float64", "shape": [-1]}}]'
  params: null
utc_time_created: '2024-03-13 08:15:55.707239'

注册

MLflow注册表是一个扩展，有助于：

• 管理每个MLModel的版本
• 记录每个模型在三个不同阶段的发展进程：归档（archive）、模拟环境(staging)和生产（production）。它非常类似于Git中的版本系统。

注册模型有四种方式：

1、通过UI
2、作为“MLflow.<flavor>.log_model()”的参数方式
3、使用“MLflow.register_model()”方法或
4、使用“create_registered_model()”客户端API。

使用“MLflow.<flavor>.log_model()”方法注册模型：

with MLflow.start_run():

   model = DecisionTreeModel(max_depth=max_depth)
   model.load_data()
   model.train()
   model.evaluate()

   MLflow.log_param("tree_depth", max_depth)
   MLflow.log_metric("precision", model.precision)
   MLflow.log_metric("recall", model.recall)
   MLflow.log_metric("accuracy", model.accuracy)

   # Register the model 比model 保存多了个registered_model_name参数
   MLflow.sklearn.log_model(model.tree, "MyModel-dt",      registered_model_name="Decision Tree")

如果是新模型，MLFlow将其初始化为版本1。如果模型已进行版本控制，则将其初始化成版本2（或后续版本）。

默认情况下，注册模型时，分配的状态为“无”。要将状态分配给已注册模型，可以通过以下方式执行

client = MLflowClient()
client.transition_model_version_stage(
    name="Decision Tree",
    version=2,
    stage="Staging"
)

在上面的代码片段中，版本2的模型被分配给模拟环境(staging)。

image.png

使用MLflowCLI实现模型服务。需要服务器URI、模型名称和模型状态这些信息即可，如下所示：

$ export MLflow_TRACKING_URI=http://10.0.102.50:15000
$ mlflow models serve -m "models:/model-v1/Staging"

Mlflow Recipes

核心概念

Templates:

一个mlflow工作流编排的工程模板目录，里面包含了 配置文件模板recipes.yaml , profile，执行步骤的代码等数据文件。用户开发一个mlflow工作流可以参考templates的目录结构进行修改，官方提供了一个mlflow recipes模板，参考github地址 recipes-regression-template。

模板工程的目录组成：

recipes-regression-template
|-- LICENSE
|-- notebooks               # 工作流编排文件，支持python和java两种api
|   |-- databricks.py
|   `-- jupyter.ipynb
|-- profiles
|   |-- databricks.yaml     # 存放profile实例，运行一个工作流时选择一个profile来填充 recipes.yaml 模板配置文件的参数
|   `-- local.yaml
|-- README.md
|-- recipe.yaml             # 模板配置文件，定义recipes的基本信息： steps，name等
|-- requirements            
|   |-- lint-requirements.txt
|   `-- test-requirements.txt
|-- requirements.txt        # 运行一个recipes需要的 环境依赖信息
|-- steps                   # 工作流步骤对应的代码存放处
|   |-- custom_metrics.py
|   |-- ingest.py
|   |-- split.py
|   |-- train.py
|   `-- transform.py
`-- tests                   # 单元测试
    |-- __init__.py
    |-- train_test.py
     `-- transform_test.py

Recipes:

编排工作流的核心，提供了python和java两种编程语言的SDK。开发者通过编写代码实现一套工作流程，因此十分容易拓展。编排代码存放到notebooks目录。

# 准备环境依赖：requirements.txt
dbutils.library.restartPython()

from mlflow.recipes import Recipe
# 指定profile，填充recipes模板参数
r = Recipe(profile="databricks")
# 生成dag可视化数据供UI解析
r.inspect()

# 执行步骤 ingest
r.run("ingest")

# 执行步骤 split
r.run("split")

training_data = r.get_artifact("training_data")
training_data.describe()

# 执行步骤 transform
r.run("transform")

# 执行步骤 train
r.run("train")

trained_model = r.get_artifact("model")
print(trained_model)

# 执行步骤，评估模型
r.run("evaluate")

# 执行步骤，注册模型
r.run("register")

step

表示工作流程中的一个步骤，即DAG 中的一个节点。

Profiles

工作流配置文件 recipes.yaml 是一个模板，yaml文件可以通过占位符的方式定义参数，参数的值在运行工作流时通过profile注入，profile通过yaml格式编写。

from mlflow.recipes import Recipe
# 指定profile，填充recipes模板参数
r = Recipe(profile="databricks")

可视化

通过调用函数 Recipe.inspect() 来生成DAG数据，供UI解析