在批处理部署中部署 MLflow 模型

  • 项目

适用范围：Azure CLI ml 扩展 v2（最新版）Python SDK azure-ai-ml v2（最新版）

本文介绍如何将 MLflow 模型部署到 Azure 机器学习以使用批处理终结点进行批处理推理。 将 MLflow 模型部署到批处理终结点时，Azure 机器学习可以：

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

注意

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

关于此示例

此示例演示如何将 MLflow 模型部署到批处理终结点以执行批处理预测。 此示例使用基于 UCI 心脏病数据集的 MLflow 模型。 数据库包含 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 中继续操作

可以在以下笔记本中按照此示例进行操作。 在克隆的存储库中，打开笔记本：mlflow-for-batch-tabular.ipynb。

先决条件

在按照本文中的步骤操作之前，请确保满足以下先决条件：

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

  • Azure 机器学习工作区。 如果没有，请使用如何管理工作区一文中的步骤创建一个。

  • 确保在工作区中具有以下权限：

    • 创建/管理批处理终结点和部署：使用允许 Microsoft.MachineLearningServices/workspaces/batchEndpoints/* 的所有者角色、参与者角色或自定义角色。

    • 在工作区资源组中创建 ARM 部署：在部署工作区的资源组中使用允许 Microsoft.Resources/deployments/write 的所有者角色、参与者角色或自定义角色。

  • 需要安装以下软件才能使用 Azure 机器学习：

    Azure CLI 和ml适用于 Azure 机器学习的扩展。

    az extension add -n ml

    注意

    Azure CLI 的 ml 扩展版本 2.7 中引入了 Batch 终结点的管道组件部署。 使用 az extension update --name ml 获取它的上一个版本。

连接到工作区

工作区是 Azure 机器学习的顶级资源，为使用 Azure 机器学习时创建的所有项目提供了一个集中的处理位置。 在本部分，我们将连接到要在其中执行部署任务的工作区。

在以下代码中传入订阅 ID、工作区、位置和资源组的值：

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

步骤

按照以下步骤将 MLflow 模型部署到批处理终结点，以便通过新数据运行批处理推理：

  1. 批处理终结点只能部署已注册的模型。 在这种情况下，我们已在存储库中拥有模型的本地副本，因此我们只需要将模型发布到工作区中的注册表。 如果打算部署的模型已注册，则可以跳过此步骤。

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

  2. 在继续操作之前，我们需要确保要创建的批处理部署可以在某些基础结构（计算）上运行。 批处理部署可以在工作区中已存在的任何 Azure 机器学习计算上运行。 这意味着多个批处理部署可以共享相同的计算基础结构。 在此示例中，我们将在名为 cpu-cluster 的 Azure 机器学习计算群集上工作。 让我们验证工作区上是否存在计算，如果不存在，则创建计算。

    按如下所示创建计算群集：

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

  3. 现在可以创建批处理终结点和部署了。 让我们先从终结点开始。 终结点只需要创建名称和说明。 终结点名称最终将包含在与终结点关联的 URI 中。 因此，批处理终结点名称在 Azure 区域内需是唯一的。 例如，chinaeast2 中只能有一个名为 mybatchendpoint 的批处理终结点。

    在本例中，我们将终结点名称放在一个变量中，以便稍后可以轻松引用它。

    ENDPOINT_NAME="heart-classifier"

  4. 创建终结点：

    如需创建新终结点，请创建如下所示的 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

    然后，用以下命令创建终结点：

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

  5. 现在，创建部署。 创建部署时，MLflow 模型不需要指示环境或评分脚本，因为它是为你创建的。 但是，如果要自定义部署的推理方式，则可以指定它们。

    如需在创建的终结点下创建新部署，请创建如下所示的 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

    然后，用以下命令创建部署：

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

    重要

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

  6. 尽管可以在终结点内调用特定部署，但通常需要调用终结点本身，让终结点决定使用哪个部署。 此类部署命名为“默认”部署。 这样可以更改默认部署，从而更改为部署提供服务的模型，且无需更改与调用终结点的用户的协定。 使用以下说明更新默认部署：

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

  7. 此时，我们的批处理终结点随时可用。

测试部署

为测试终结点，我们将使用此存储库中的未标记数据示例，此示例可与模型一起使用。 批处理终结点只能处理云中可从 Azure 机器学习工作区访问的数据。 在该示例中，我们会将其上传到 Azure 机器学习数据存储中。 具体地说，我们将创建可用于调用终结点进行评分的数据资产。 但是，请注意，批处理终结点接受可存放在多种类型的位置的数据。

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

    a. 在 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

    b. 创建数据资产：

    az ml data create -f heart-dataset-unlabeled.yml

  2. 数据已上传且随时可用，接下来我们将调用终结点：

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

    注意

    并不是每次安装时都会安装实用工具 jq。 可点击此链接了解安装说明。

    提示

    请注意，我们不会在调用操作中给出部署名称。 因为终结点会自动将作业路由到默认部署。 由于终结点只有一个部署，因此该部署为默认部署。 可通过指明自变量/参数 deployment_name 来将特定部署作为目标。

  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"]
)

输出如下所示：

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 格式的表格数据，并且有 4 个不同的输入文件（heart-unlabeled-0.csv、heart-unlabeled-1.csv、heart-unlabeled-2.csv 和 heart-unlabeled-3.csv）。

部署到批处理推理时的注意事项

Azure 机器学习支持将 MLflow 模型部署到批处理终结点，而无需指示评分脚本。 这是一种部署需要批量处理大量数据的模型的便捷方法。 Azure 机器学习使用 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 模型的注意事项。

警告

请注意，输入数据中可能存在的任何不受支持的文件都会使作业失败。 你将看到错误条目，如下所示："ERROR:azureml:处理输入文件时出错: '/mnt/batch/tasks/.../a-given-file.avro'。不支持文件类型 'avro'。"。

MLflow 模型的签名强制

输入的数据类型由批处理部署作业强制实施，同时使用可用的 MLflow 模型签名读取数据。 这意味着数据输入应符合模型签名中指示的类型。 如果无法按预期分析数据，作业将失败并显示类似于以下内容的错误消息："ERROR:azureml:Error processing input file: '/mnt/batch/tasks/.../a-given-file.csv'. Exception: invalid literal for int() with base 10: 'value'"。

提示

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

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

风格支持

批处理部署仅支持使用 pyfunc 风格部署 MLflow 模型。 如果需要部署不同的风格，请参阅将 MLflow 模型与评分脚本配合使用。

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

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

你通常会在以下情况下选择此工作流：

  • 需要处理批处理部署 MLflow 部署不支持的文件类型。
  • 需要自定义模型的运行方式，例如，使用特定风格将其与 mlflow.<flavor>.load() 一起加载。
  • 当模型本身未完成时，需要在评分例程中执行预/后处理。
  • 模型的输出不能在表格数据中很好地表示。 例如，它是表示映像的张量。
  • 由于内存限制，模型无法一次处理所有文件，它需要以区块形式进行读取。

重要

如果选择指示 MLflow 模型部署的评分脚本，则还必须指定部署将运行的环境。

步骤

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

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

    a. 转到 Azure 机器学习工作室门户。

    b. 转到“模型”部分。

    c. 选择要尝试部署的模型，然后单击“工件”选项卡。

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

    屏幕截图显示放置模型项目的文件夹。

  2. 创建评分脚本。 请注意将之前标识的文件夹名称 model 包含在 init() 函数中的方式。

    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

  4. 此时，我们的批处理终结点随时可用。

清理资源

运行以下代码以删除批处理终结点和所有基础部署。 不会删除批量评分作业。

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

后续步骤