在 Azure Machine Learning 中的批量部署中部署 MLflow 模型

适用于：Azure CLI ml 扩展 v2 （当前版本）Python SDK azure-ai-ml v2 （当前版本）

本文介绍如何使用批处理终结点将 MLflow 模型部署到Azure Machine Learning进行批处理推理。 将 MLflow 模型部署到批处理终结点时，Azure Machine Learning完成以下任务：

• 提供一个 MLflow 基础映像或特选环境，其中包含运行Machine Learning批处理作业所需的依赖项。
• 为您创建一个批处理作业管道，包含评分脚本，该管道可以通过并行化来处理数据。

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

先决条件

• 一个 Azure 订阅。 如果没有Azure订阅，请在开始前创建 Trial。

• Azure Machine Learning工作区。 若要创建工作区，请参阅 Manage Azure Machine Learning 工作区。

• Azure Machine Learning工作区中的以下权限：

  • 若要创建或管理批处理终结点和部署：使用具有Microsoft.MachineLearningServices/workspaces/batchEndpoints/*权限的所有者、参与者或自定义角色。
  • 若要在工作区资源组中创建Azure Resource Manager部署：请在部署工作区的资源组中使用具有Microsoft.Resources/deployments/write权限的所有者、参与者或自定义角色。

• 用于Python的 Azure Machine Learning CLI 或 Azure Machine Learning SDK：

  • Azure CLI
  • Python

  运行以下命令安装 Azure CLI 和 mlextension for Azure Machine Learning：

  az extension add -n ml
  

  批处理终结点的管道组件部署需要Azure CLI的 ml 扩展版本 2.7 或更高版本（当前版本：2.37.0）。 使用 az extension update --name ml 命令获取最新版本。

连接到工作区

工作区是Azure Machine Learning的顶级资源。 它提供了一个集中的位置，用于处理使用Azure Machine Learning时创建的所有项目。 在本部分，你将连接到要在其中执行部署任务的工作区。

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

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

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

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

请在 Jupyter Notebooks 中跟随学习

您可以使用公共 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"

model_name = "heart-classifier-mlflow"
model_local_path = "model"

model = ml_client.models.create_or_update(
	Model(name=model_name, path=model_local_path, type=AssetTypes.MLFLOW_MODEL)
)

创建计算群集

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

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

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

compute_name = "batch-cluster"
if not any(filter(lambda m: m.name == compute_name, ml_client.compute.list())):
	compute_cluster = AmlCompute(
    name=compute_name, description="amlcompute", min_instances=0, max_instances=5
)
	ml_client.begin_create_or_update(compute_cluster).result()

创建批处理终结点

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

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

   • Azure CLI
   • Python

   运行以下命令：

   ENDPOINT_NAME="heart-classifier"
   

2. 创建终结点：

   • Azure CLI
   • Python

   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 模型不需要指定环境或评分脚本。 系统将自动为你创建环境或评分脚本。 但是，如果你要自定义部署执行推理的方式，则可以指定环境或评分脚本。

调用端点

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

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

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

endpoint = ml_client.batch_endpoints.get(endpoint.name)
endpoint.defaults.deployment_name = deployment.name
ml_client.batch_endpoints.begin_create_or_update(endpoint).result()

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

测试部署

若要测试终结点，请使用此存储库中可用于模型的未标记数据示例。 Batch 终结点只能处理位于云中的数据，并且可从Machine Learning工作区访问。 在此示例中，将示例上传到Machine Learning数据存储。 您创建一个用于调用终结点进行评分的数据资产。 请记住，批处理终结点接受可在各种位置放置的数据。

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

   • Azure CLI
   • Python

   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 将特定部署指定为目标。

   • Azure CLI
   • Python

   运行以下命令：

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

   注意

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

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

   • Azure CLI
   • Python

   运行以下命令：

   az ml job show -n $JOB_NAME --web
   

分析输出

部署配置指定作业在 predictions.csv 文件中生成输出预测。 该作业将创建一个命名 分数 的输出，并将此文件放在此位置。 每个批处理作业只生成一个文件。

该文件具有以下结构：

• 发送到模型的每个数据点对应一行。 对于表格数据， predictions.csv 文件包含每个已处理文件中存在的每一行的一行。 对于其他数据类型（图像、音频、文本），每个已处理的文件都有一行。

• 该文件按指定顺序包含以下字段：

  • row（可选）：输入数据文件中的相应行索引。 仅当输入数据为表格格式时，此列才适用。 预测将按照它们在输入文件中的出现顺序返回。 可以按照行号来匹配相应的预测。

  • prediction：与输入数据关联的预测。 模型通过其 predict() 函数返回此值“as-is”。

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

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

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

ml_client.jobs.download(name=scoring_job.name, download_path=".", output_name="score")

下载文件后，可以使用你偏好的编辑工具将其打开。 以下示例使用 pandas DataFrame 加载预测。

import pandas as pd

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

输出将显示一个表格：

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

Azure Machine Learning支持将 MLflow 模型部署到批处理终结点，而无需指示评分脚本。 使用这种方法可以方便地部署需要处理大量数据（类似于批处理）的模型。 Azure Machine Learning使用 MLflow 模型规范中的信息来协调推理过程。

探索工人中的工作分配

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

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

查看支持的文件类型

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

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

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

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

查看变体支持

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

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

可以将 MLflow 模型部署到批处理终结点，而无需在部署定义中指示评分脚本。 但是，你可以选择指定此文件（通常称为批处理驱动程序）来自定义推理执行。

通常，为以下方案选择此工作流：

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

使用评分脚本

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

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

   1. 在 Azure Machine Learning studio 中，浏览到 Models。

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

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

      

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 软件包。

   注意

   azureml-core 包是自 2025 年 3 月 31 日弃用的 Azure Machine Learning SDK v1 的一部分，2026 年 6 月 30 日终止支持。 它作为批处理部署评分脚本的运行时依赖项仍是必需的。

   提示

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

   重要

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

   • Azure CLI
   • Python

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

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

2. 配置部署：

   • Azure CLI
   • Python

   若要在创建的终结点下创建新部署，请如以下代码片段所示创建 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. 创建部署：

   • Azure CLI
   • Python

   运行以下代码：

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

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

清理资源

完成练习后，删除不再需要的资源。

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

ml_client.batch_endpoints.begin_delete(endpoint_name).result()

相关内容

• 查看有关部署到批量推理的注意事项
• 自定义批量部署中的输出