使用 Azure 机器学习在批处理部署中部署 MLflow 模型

适用范围:Azure CLI ml 扩展 v2(最新版)Python SDK azure-ai-ml v2(最新版)

本文介绍如何将 MLflow 模型部署到 Azure 机器学习,以使用批处理终结点进行批量推理。 将 MLflow 模型部署到批处理终结点时,Azure 机器学习将完成以下任务:

  • 提供 MLflow 基础映像或策展环境,其中包含运行机器学习批处理作业所需的依赖项。
  • 创建包含评分脚本的批处理作业管道,该脚本可用于使用并行化处理数据。

有关支持的输入文件类型的详细信息,以及有关 MLflow 模型工作原理的详细信息,请参阅有关部署到批量推理的注意事项

先决条件

  • Azure 订阅。 如果没有 Azure 订阅,可在开始前创建一个试用帐户。 试用 Azure 机器学习

  • Azure 机器学习工作区。 若要创建工作区,请参阅管理 Azure 机器学习工作区

  • 确保你在机器学习工作区中拥有以下权限:

    • 创建或管理批处理终结点和部署:使用允许 Microsoft.MachineLearningServices/workspaces/batchEndpoints/* 的所有者、参与者或自定义角色。
    • 在工作区资源组中创建 Azure 资源管理器部署:在部署工作区的资源组中使用允许 Microsoft.Resources/deployments/write 的所有者、参与者或自定义角色。
  • 安装以下软件以使用机器学习:

    运行以下命令,以安装 Azure CLIAzure 机器学习的 ml 扩展

    az extension add -n ml
    

    Azure CLI 的 ml 扩展版本 2.7 中引入了批处理终结点的管道组件部署。 使用 az extension update --name ml 命令获取最新版本。


连接到工作区

工作区是机器学习的顶级资源。 它提供一个集中的位置用于处理使用机器学习时创建的所有工件。 在本部分,你将连接到要在其中执行部署任务的工作区。

在以下命令中,请输入你的订阅 ID、工作区、位置和资源组的值:

az account set --subscription <subscription>
az configure --defaults workspace=<workspace> group=<resource-group> location=<location>

浏览示例

本文中的示例演示如何将 MLflow 模型部署到批处理终结点以执行批量预测。 该 MLflow 模型基于 UCI 心脏病数据集。 数据库包含 76 个属性,但本示例仅使用其中的 14 个属性。 该模型尝试根据从 0(有心脏病)到 1(没有心脏病)的整数值来预测患者是否有心脏病。

已使用 XGBBoost 分类器训练该模型。 所有必要的预处理操作都已打包为 scikit-learn 管道,因此该模型可以充当一个端到端的管道来完成从原始数据处理到做出预测的任务。

本文中的示例基于 azureml-examples 存储库中包含的代码示例。 要在无需复制/粘贴 YAML 和其他文件的情况下在本地运行命令,请先克隆存储库,然后将目录更改为以下文件夹:

git clone https://github.com/Azure/azureml-examples --depth 1
cd azureml-examples/cli

本示例的文件位于以下文件夹中:

cd endpoints/batch/deploy-models/heart-classifier-mlflow

在 Jupyter Notebook 中继续操作

可以使用公共 Jupyter Notebook 来学习本示例。 在克隆的存储库中,打开 mlflow-for-batch-tabular.ipynb 笔记本。

部署 MLflow 模型

在本部分,你会将 MLflow 模型部署到批处理终结点,以便可以基于新数据运行批量推理。 在继续部署之前,需要确保你的模型已注册,并且工作区中有一个可用的计算群集。

注册模型

批处理终结点只能部署已注册的模型。 在本文中,你将使用存储库中模型的本地副本。 因此,只需将模型发布到工作区中的注册表即可。

注意

如果要部署的模型已注册,则可以转到创建计算群集部分。

运行以下命令注册模型:

MODEL_NAME='heart-classifier-mlflow'
az ml model create --name $MODEL_NAME --type "mlflow_model" --path "model"

创建计算群集

需要确保批处理部署可以在某些可用的基础结构(计算)上运行。 批处理部署可以在工作区中已存在的任何机器学习计算上运行。 多个批处理部署可以共享同一计算基础结构。

在本文中,你将使用名为 cpu-cluster 的机器学习计算群集。 以下示例将验证工作区中是否存在计算,或创建新的计算。

创建计算群集:

az ml compute create -n batch-cluster --type amlcompute --min-instances 0 --max-instances 5

创建批处理终结点

若要创建终结点,需要指定名称和说明。 终结点名称将显示在与终结点关联的 URI 中,因此在 Azure 区域中必须是唯一的。 例如,在 WestUS2 区域中,只能存在一个名为 mybatchendpoint 的批处理终结点。

  1. 将终结点名称放在变量中,以便稍后轻松引用:

    运行以下命令:

    ENDPOINT_NAME="heart-classifier"
    
  2. 创建终结点:

    1. 若要创建新的终结点,请如以下代码所示创建 YAML 配置:

      endpoint.yml

      $schema: https://azuremlschemas.azureedge.net/latest/batchEndpoint.schema.json
      name: heart-classifier-batch
      description: A heart condition classifier for batch inference
      auth_mode: aad_token
      
    2. 使用以下命令创建终结点:

      az ml batch-endpoint create -n $ENDPOINT_NAME -f endpoint.yml
      

创建批处理部署

当你创建部署时,MLflow 模型不要求你指定环境或评分脚本。 系统将自动为你创建环境或评分脚本。 但是,如果你要自定义部署执行推理的方式,则可以指定环境或评分脚本。

  1. 若要在创建的终结点下创建新部署,请如以下代码所示创建 YAML 配置。 可以检查额外属性中的完整批处理终结点 YAML 机构

    deployment-simple/deployment.yml

    $schema: https://azuremlschemas.azureedge.net/latest/modelBatchDeployment.schema.json
    endpoint_name: heart-classifier-batch
    name: classifier-xgboost-mlflow
    description: A heart condition classifier based on XGBoost
    type: model
    model: azureml:heart-classifier-mlflow@latest
    compute: azureml:batch-cluster
    resources:
      instance_count: 2
    settings:
      max_concurrency_per_instance: 2
      mini_batch_size: 2
      output_action: append_row
      output_file_name: predictions.csv
      retry_settings:
        max_retries: 3
        timeout: 300
      error_threshold: -1
      logging_level: info
    
  2. 使用以下命令创建部署:

    az ml batch-deployment create --file deployment-simple/deployment.yml --endpoint-name $ENDPOINT_NAME --set-default
    

重要

根据模型在单个批上运行推理所需的时间,在部署中配置 timeout 值。 批大小越大,timeout 值越长。 请记住,mini_batch_size 值表示批中的文件数,而不是样本数。 使用表格数据时,每个文件可能包含多个行,这将增加批处理终结点处理每个文件所需的时间。 在这种情况下,请使用更大的 timeout 值来避免超时错误。

调用终结点

尽管可以在终结点内部调用特定部署,但通常需要调用终结点本身,并支持终结点决定使用哪个部署。 此类部署称为“默认”部署。 这种方法允许更改默认部署,因而可以更改处理部署的模型,而无需更改与调用终结点的用户之间的协定。

使用以下说明更新默认部署:

DEPLOYMENT_NAME="classifier-xgboost-mlflow"
az ml batch-endpoint update --name $ENDPOINT_NAME --set defaults.deployment_name=$DEPLOYMENT_NAME

该批处理终结点现在可供使用。

测试部署

若要测试终结点,请使用此存储库中可与模型配合使用的未标记数据示例。 批处理终结点只能处理云中可从机器学习工作区访问的数据。 本示例将示例数据上传到机器学习数据存储中。 你将创建可用于调用终结点进行评分的数据资产。 请注意,批处理终结点接受可存放在多种类型的位置的数据。

  1. 首先创建数据资产。 该数据资产由包含多个 CSV 文件的文件夹组成,我们希望使用批处理终结点并行处理这些文件。 如果数据已注册为数据资产,或者想要使用不同的输入类型,则可以跳过此步骤。

    1. 在 YAML 中创建数据资产定义:

      heart-dataset-unlabeled.yml

      $schema: https://azuremlschemas.azureedge.net/latest/data.schema.json
      name: heart-dataset-unlabeled
      description: An unlabeled dataset for heart classification.
      type: uri_folder
      path: data
      
    2. 创建数据资产:

      az ml data create -f heart-dataset-unlabeled.yml
      
  2. 上传数据后,调用终结点。

    提示

    在以下命令中,请注意 invoke 操作中未指定部署名称。 终结点会自动将作业路由到默认部署,因为终结点只有一个部署。 可通过指明自变量/参数 deployment_name 将特定部署指定为目标。

    运行以下命令:

    JOB_NAME = $(az ml batch-endpoint invoke --name $ENDPOINT_NAME --input azureml:heart-dataset-unlabeled@latest | jq -r '.name') 
    

    注意

    并不是每次安装时都会安装实用工具 jq。 有关安装说明,请参阅下载 jq

  3. 命令返回后立即启动批处理作业。 在作业完成前可监视作业状态:

    运行以下命令:

    az ml job show -n $JOB_NAME --web
    

分析输出

输出预测在 predictions.csv 文件中生成,如部署配置中所示。 该作业将生成名为 score 的输出,此文件就位于其中。 每个批处理作业仅生成一个文件。

文件的结构如下所示:

  • 会将每个数据点对应的一行发送到模型。 对于表格数据,predictions.csv 文件包含每个已处理文件中每一行对应的一行。 对于其他数据类型(图像、音频、文本),此文件包含每个已处理文件对应的一行。

  • 该文件包含以下列(按指定顺序列出):

    • row(可选):输入数据文件中的相应行索引。 仅当输入数据为表格格式时,此列才适用。 预测将按照这些列在输入文件中的显示顺序返回。 可以按照行号来匹配相应的预测。

    • prediction:与输入数据关联的预测。 此值“按原样”返回,因为它是由模型的 predict(). 函数提供的。

    • file_name:从中读取数据的文件的名称。 在表格数据中,可以使用此字段来判断哪个预测属于每项输入数据。

可以使用作业名称下载作业的结果。

如需下载预测,请使用以下命令:

az ml job download --name $JOB_NAME --output-name score --download-path ./

下载文件后,可以使用你偏好的编辑工具将其打开。 以下示例使用 Pandas 数据帧加载预测。

import pandas as pd

score = pd.read_csv(
    "named-outputs/score/predictions.csv", names=["row", "prediction", "file"]
)

输出将显示一个表格:

预测 文件
0 0 heart-unlabeled-0.csv
1 1 heart-unlabeled-0.csv
2 0 heart-unlabeled-0.csv
... ... ...
307 0 heart-unlabeled-3.csv

提示

请注意,在此示例中,输入数据包含 CSV 格式的表格数据。 有四个不同的输入文件:heart-unlabeled-0.csv、heart-unlabeled-1.csv、heart-unlabeled-2.csv 和 heart-unlabeled-3.csv

查看有关批量推理的注意事项

机器学习支持将 MLflow 模型部署到批处理终结点,而无需指定评分脚本。 使用这种方法可以方便地部署需要处理大量数据(类似于批处理)的模型。 机器学习使用 MLflow 模型规范中的信息来协调推理过程。

浏览工作器上的工作分布

对于结构化和非结构化数据,批处理终结点都在文件级别分发工作。 因此,此功能仅支持 URI 文件URI 文件夹。 每个辅助角色一次处理 Mini batch size 个文件。 对于表格数据,在分发工作时,批处理终结点不会考虑每个文件中的行数。

警告

在推理期间无法浏览嵌套的文件夹结构。 如果你使用文件夹对数据进行分区,请务必在继续之前平展结构。

批处理部署将为每个文件调用 MLflow 模型的 predict 函数一次。 对于包含多行的 CSV 文件,此操作可能会给基础计算带来内存压力。 该行为会增加模型为单个文件评分所需的时间,特别是对于大型语言模型等高开销模型。 如果日志中出现多个内存不足异常或超时条目,请考虑将数据拆分为包含较少行的较小文件,或在模型/评分脚本内的行级别实现批处理。

查看支持的文件类型

在未指定环境或评分脚本的情况下部署 MLflow 模型时,批量推理支持以下数据类型。 若要处理不同的文件类型或以不同的方式执行推理,可以通过使用评分脚本自定义 MLflow 模型部署来创建部署。

文件扩展名 作为模型输入返回的类型 签名要求
.csv.parquet.pqt pd.DataFrame ColSpec。 如果未提供,则不会强制执行列类型化。
.png.jpg.jpeg.tiff.bmp.gif np.ndarray TensorSpec。 输入将重新塑形,以匹配张量形状(如果可用)。 如果没有可用的签名,则推断类型为 np.uint8 的张量。 有关详细信息,请参阅处理图像的 MLflow 模型的注意事项

警告

在输入数据中包含任何不受支持的文件都会导致作业失败。 在这种情况下,你会看到如下所示的错误:“错误: azureml: 处理输入文件时出错:‘/mnt/batch/tasks/.../a-given-file.avro’。不支持文件类型‘avro’”。

了解 MLflow 模型的签名强制要求

批处理部署作业在使用可用的 MLflow 模型签名读取数据时,会强制要求提供输入的数据类型。 因此,数据输入应该符合模型签名中指定的类型。 如果无法按预期分析数据,则作业将会失败,并出现如下所示的错误:“错误: azureml: 处理输入文件时出错:‘/mnt/batch/tasks/.../a-given-file.csv’。异常: 以 10 为基数的 int() 的文本无效:‘value’。”

提示

MLflow 模型中的签名是可选的,但我们强烈建议使用。 它们为提前检测数据兼容性问题提供了一种便捷的方法。 有关如何使用签名记录模型的详细信息,请参阅使用自定义签名、环境或示例记录模型

可以通过打开与 MLflow 模型关联的 MLmodel 文件来检查模型的模型签名。 有关签名在 MLflow 中的工作原理的详细信息,请参阅 MLflow 中的签名

了解风格支持

批处理部署仅支持使用 pyfunc 风格部署 MLflow 模型。 若要部署其他风格,请参阅使用评分脚本自定义模型部署

使用评分脚本自定义模型部署

可以在不指示部署定义中的评分脚本的情况下将 MLflow 模型部署到批处理终结点。 但是,你可以选择指定此文件(通常称为批处理驱动程序)来自定义推理执行

对于以下方案,你通常会选择此工作流:

  • 处理 MLflow 部署的批处理部署不支持的文件类型。
  • 自定义模型的运行方式,例如使用特定的风格通过 mlflow.<flavor>.load() 函数加载模型。
  • 在评分例程中完成模型本身未完成的预处理或后处理。
  • 调整无法使用表格数据很好地呈现的模型(例如表示图像的张量图)的呈现方式。
  • 允许模型逐区块读取数据,因为内存限制导致模型无法一次性处理每个文件。

重要

若要为 MLflow 模型部署指定评分脚本,需要指定运行部署的环境。

使用评分脚本

使用以下步骤通过自定义评分脚本部署 MLflow 模型:

  1. 标识放置 MLflow 模型的文件夹。

    1. Azure 机器学习门户中,浏览到“模型”

    2. 选择要部署的模型,然后选择“工件”选项卡

    3. 记下显示的文件夹。 注册模型时指明了此文件夹。

      显示模型工件所在的文件夹的屏幕截图。

  2. 创建评分脚本。 请注意上述文件夹名称 modelinit() 函数中的显示方式。

    deployment-custom/code/batch_driver.py

# Copyright (c) Microsoft. All rights reserved.
# Licensed under the MIT license.

import os
import glob
import mlflow
import pandas as pd
import logging


def init():
    global model
    global model_input_types
    global model_output_names

    # AZUREML_MODEL_DIR is an environment variable created during deployment
    # It is the path to the model folder
    # Please provide your model's folder name if there's one
    model_path = glob.glob(os.environ["AZUREML_MODEL_DIR"] + "/*/")[0]

    # Load the model, it's input types and output names
    model = mlflow.pyfunc.load(model_path)
    if model.metadata and model.metadata.signature:
        if model.metadata.signature.inputs:
            model_input_types = dict(
                zip(
                    model.metadata.signature.inputs.input_names(),
                    model.metadata.signature.inputs.pandas_types(),
                )
            )
        if model.metadata.signature.outputs:
            if model.metadata.signature.outputs.has_input_names():
                model_output_names = model.metadata.signature.outputs.input_names()
            elif len(model.metadata.signature.outputs.input_names()) == 1:
                model_output_names = ["prediction"]
    else:
        logging.warning(
            "Model doesn't contain a signature. Input data types won't be enforced."
        )


def run(mini_batch):
    print(f"run method start: {__file__}, run({len(mini_batch)} files)")

    data = pd.concat(
        map(
            lambda fp: pd.read_csv(fp).assign(filename=os.path.basename(fp)), mini_batch
        )
    )

    if model_input_types:
        data = data.astype(model_input_types)

    # Predict over the input data, minus the column filename which is not part of the model.
    pred = model.predict(data.drop("filename", axis=1))

    if pred is not pd.DataFrame:
        if not model_output_names:
            model_output_names = ["pred_col" + str(i) for i in range(pred.shape[1])]
        pred = pd.DataFrame(pred, columns=model_output_names)

    return pd.concat([data, pred], axis=1)

  1. 创建一个可以执行评分脚本的环境。 由于本示例中的模型是 MLflow,因此模型包中还指定了 conda 要求。 有关 MLflow 模型和包含的文件的详细信息,请参阅 MLmodel 格式

    在此步骤中,你将使用文件中的 conda 依赖项来生成环境。 此外,需要包含批处理部署所需的 azureml-core 包。

    提示

    如果你的模型已在模型注册表中注册,则可以下载并复制与该模型关联的 conda.yml 文件。 在 Azure 机器学习工作室中的“模型”>“从列表中选择你的模型”>“工件”下可以找到该文件。 在根文件夹中选择 conda.yml 文件,然后选择“下载”或复制其内容

    重要

    此示例使用在 /heart-classifier-mlflow/environment/conda.yaml 中指定的 Conda 环境。 此文件是通过合并原始 MLflow conda 依赖项文件并添加 azureml-core 包创建的。 不能直接从模型使用 conda.yml 文件。

    环境定义将作为匿名环境包含在部署定义本身中。 在部署中可以看到以下几行:

    environment:
      name: batch-mlflow-xgboost
      image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04:latest
      conda_file: environment/conda.yaml
    
  2. 配置部署:

    若要在创建的终结点下创建新部署,请如以下代码片段所示创建 YAML 配置。 可以检查额外属性中的完整批处理终结点 YAML 机构

    deployment-custom/deployment.yml

    $schema: https://azuremlschemas.azureedge.net/latest/batchDeployment.schema.json
    endpoint_name: heart-classifier-batch
    name: classifier-xgboost-custom
    description: A heart condition classifier based on XGBoost
    type: model
    model: azureml:heart-classifier-mlflow@latest
    environment:
    name: batch-mlflow-xgboost
    image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04:latest
    conda_file: environment/conda.yaml
    code_configuration:
    code: code
    scoring_script: batch_driver.py
    compute: azureml:batch-cluster
    resources:
    instance_count: 2
    settings:
    max_concurrency_per_instance: 2
    mini_batch_size: 2
    output_action: append_row
    output_file_name: predictions.csv
    retry_settings:
      max_retries: 3
      timeout: 300
    error_threshold: -1
    logging_level: info
    
  3. 创建部署:

    运行以下代码:

    az ml batch-deployment create --file deployment-custom/deployment.yml --endpoint-name $ENDPOINT_NAME
    

该批处理终结点现在可供使用。

清理资源

完成本练习后,请删除不再需要的资源。

运行以下代码以删除批处理终结点和所有基础部署:

az ml batch-endpoint delete --name $ENDPOINT_NAME --yes

此命令不会删除批处理评分作业。