你当前正在访问 Microsoft Azure Global Edition 技术文档网站。如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

如何部署管道以使用预处理执行批量评分

项目
09/03/2024

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

在本文中，你将了解如何在批处理终结点下部署推理（或评分）管道。管道可对已注册的模型执行评分，同时从训练模型时重用预处理组件。重用相同的预处理组件可确保在评分期间应用相同的预处理。

你将了解以下内容：

创建可重用工作区中现有组件的管道
将管道部署到终结点
使用管道生成的预测

关于此示例

此示例演示了如何在使用模型进行推理之前重用预处理代码和在预处理过程中学习的参数。通过重用预处理代码和学习的参数，我们可以确保在推理过程中也会应用已在训练期间应用于输入数据的同一转换（例如规范化和特征编码）。用于推理的模型将对 UCI 心脏病数据集中的表格数据执行预测。

管道的可视化效果如下所示：

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

Azure CLI
Python

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-pipelines/batch-scoring-with-preprocessing

在 Jupyter Notebook 中继续操作

可以通过在克隆的存储库中打开 sdk-deploy-and-test.ipynb 笔记本来遵循此示例的 Python SDK 版本。

先决条件

Azure 订阅。如果没有 Azure 订阅，请在开始操作前先创建一个免费帐户。试用免费版或付费版 Azure 机器学习。
一个 Azure 机器学习工作区。若要创建工作区，请参阅管理 Azure 机器学习工作区。
确保你在机器学习工作区中拥有以下权限：
- 创建或管理批处理终结点和部署：使用允许 Microsoft.MachineLearningServices/workspaces/batchEndpoints/* 的所有者、参与者或自定义角色。
- 在工作区资源组中创建 Azure 资源管理器部署：在部署工作区的资源组中使用允许 Microsoft.Resources/deployments/write 的所有者、参与者或自定义角色。
安装以下软件以使用机器学习：
- Azure CLI
- Python
运行以下命令，以安装 Azure CLI 和 Azure 机器学习的 ml 扩展：
```
az extension add -n ml
```
Azure CLI 的 ml 扩展版本 2.7 中引入了批处理终结点的管道组件部署。使用 az extension update --name ml 命令获取最新版本。
运行以下命令安装适用于 Python 的 Azure 机器学习 SDK：
```
pip install azure-ai-ml
```
该 SDK 的 1.7.0 版本中引入了 ModelBatchDeployment 和 PipelineComponentBatchDeployment 类。使用 pip install -U azure-ai-ml 命令获取最新版本。

连接到工作区

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

Azure CLI
Python

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

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

导入所需的库：

from azure.ai.ml import MLClient, Input, load_component
from azure.ai.ml.entities import BatchEndpoint, ModelBatchDeployment, ModelBatchDeploymentSettings, PipelineComponentBatchDeployment, Model, AmlCompute, Data, BatchRetrySettings, CodeConfiguration, Environment, Data
from azure.ai.ml.constants import AssetTypes, BatchDeploymentOutputAction
from azure.ai.ml.dsl import pipeline
from azure.identity import DefaultAzureCredential

配置工作区详细信息并获取工作区句柄：

在以下命令中，请输入你的订阅 ID、工作区和资源组的值：

subscription_id = "<subscription>"
resource_group = "<resource-group>"
workspace = "<workspace>"

ml_client = MLClient(DefaultAzureCredential(), subscription_id, resource_group, workspace)

创建推理管道

在本部分中，我们将创建推理管道所需的所有资产。我们将首先创建一个环境，其中包含管道组件所需的库。接下来，创建运行批处理部署的计算群集。之后，注册构建推理管道所需的组件、模型和转换。最后，生成并测试管道。

创建环境

此示例中的组件将使用包含 XGBoost 和 scikit-learn 库的环境。 environment/conda.yml 文件包含环境的配置：

environment/conda.yml

channels:
- conda-forge
dependencies:
- python=3.8.5
- pip
- pip:
  - mlflow
  - azureml-mlflow
  - datasets
  - jobtools
  - cloudpickle==1.6.0
  - dask==2023.2.0
  - scikit-learn==1.1.2
  - xgboost==1.3.3
name: mlflow-env

按如下所示创建环境：

定义环境：

Azure CLI
Python

environment/xgboost-sklearn-py38.yml

$schema: https://azuremlschemas.azureedge.net/latest/environment.schema.json
name: xgboost-sklearn-py38
image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04:latest
conda_file: conda.yml
description: An environment for models built with XGBoost and Scikit-learn.

environment = Environment(
    name="xgboost-sklearn-py38",
    description="An environment for models built with XGBoost and Scikit-learn.",
    image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04:latest",
    conda_file="environment/conda.yml",
)

创建环境：

Azure CLI
Python

az ml environment create -f environment/xgboost-sklearn-py38.yml

try:
    ml_client.environments.create_or_update(environment)
except ResourceExistsError:
    pass

创建计算群集

批处理终结点和部署在计算群集上运行。它们可以运行在工作区中已存在的任何 Azure 机器学习计算群集上。因此，多个批处理部署可以共享相同的计算基础结构。在此示例中，我们将在名为 batch-cluster 的 Azure 机器学习计算群集上工作。让我们验证工作区上是否存在计算，如果不存在，则创建计算。

Azure CLI
Python

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="Batch endpoints compute cluster",
        min_instances=0,
        max_instances=5,
    )
    ml_client.begin_create_or_update(compute_cluster).result()

注册组件和模型

我们将注册构建推理管道所需的组件、模型和转换。我们可以重用其中一些资产来训练例程。

提示

在本教程中，我们将重用早期训练管道中的模型和预处理组件。可以按照示例如何使用批处理终结点部署训练管道来了解它们的创建方式。

注册用于预测的模型：

Azure CLI
Python

az ml model create --name heart-classifier --type mlflow_model --path model

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

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

已注册模型不是直接根据输入数据训练的。而是在训练之前使用准备组件对输入数据进行预处理（或转换）。我们还需要注册此组件。注册准备组件：
- Azure CLI
- Python
```
az ml component create -f components/prepare/prepare.yml
```
```
prepare_data = load_component(source="components/prepare/prepare.yml")

ml_client.components.create_or_update(prepare_data)
```
提示

注册准备组件后，现在可以从工作区中引用它。例如，azureml:uci_heart_prepare@latest 将获取准备组件的最后一个版本。

作为准备组件中数据转换的一部分，输入数据已经过规范化处理，以将预测器居中，并将其值限制在 [-1, 1] 范围内。转换参数是在 scikit-learn 转换中捕获的，我们也可以在以后有了新数据时注册并应用该转换。按如下所示注册转换：

Azure CLI
Python

az ml model create --name heart-classifier-transforms --type custom_model --path transformations

transformation_name = "heart-classifier-transforms"
transformation_local_path = "transformations"

transformations = ml_client.models.create_or_update(
    Model(
        name=transformation_name,
        path=transformation_local_path,
        type=AssetTypes.CUSTOM_MODEL,
    )
)

我们将使用另一个对给定模型计算预测的组件（名为 score）对已注册模型执行推理。我们将直接从其定义中引用组件。

提示

最佳做法是注册组件并从管道引用它。但在此示例中，我们将直接从其定义中引用组件，以帮助你查看从训练管道重用的组件以及哪些组件是新的组件。

构建管道

现在，可以将所有元素绑定在一起。我们将要部署的推理管道包括两个组件（步骤）：

preprocess_job：此步骤将读取输入数据并返回准备好的数据和已应用的转换。该步骤会接收两个输入：
- data：包含要评分的输入数据的文件夹
- transformations：（可选）将要应用的转换路径（如果可用）。提供转换后，将从路径上指示的模型中读取转换。但如果未提供路径，则将从输入数据中学习转换。但是，对于推理，无法从输入数据中学习转换参数（在本示例中为规范化系数），因为你需要使用在训练期间学习的相同参数值。由于此输入是可选的，因此可以在训练和评分期间使用 preprocess_job 组件。
score_job：此步骤将使用输入模型对转换后的数据执行推理。请注意，组件将使用 MLflow 模型来执行推理。最后，将以读取分数时的格式写回分数。

Azure CLI
Python

管道配置是在 pipeline.yml 文件中定义的：

pipeline.yml

$schema: https://azuremlschemas.azureedge.net/latest/pipelineComponent.schema.json
type: pipeline

name: batch_scoring_uci_heart
display_name: Batch Scoring for UCI heart
description: This pipeline demonstrates how to make batch inference using a model from the Heart Disease Data Set problem, where pre and post processing is required as steps. The pre and post processing steps can be components reusable from the training pipeline.

inputs:
  input_data:
    type: uri_folder
  score_mode:
    type: string
    default: append

outputs: 
  scores:
    type: uri_folder
    mode: upload

jobs:
  preprocess_job:
    type: command
    component: azureml:uci_heart_prepare@latest
    inputs:
      data: ${{parent.inputs.input_data}}
      transformations: 
        path: azureml:heart-classifier-transforms@latest
        type: custom_model
    outputs:
      prepared_data:
  
  score_job:
    type: command
    component: components/score/score.yml
    inputs:
      data: ${{parent.jobs.preprocess_job.outputs.prepared_data}}
      model:
        path: azureml:heart-classifier@latest
        type: mlflow_model
      score_mode: ${{parent.inputs.score_mode}}
    outputs:
      scores: 
        mode: upload
        path: ${{parent.outputs.scores}}

prepare_data = ml_client.components.get("uci_heart_prepare", label="latest")
score_data = load_component(source="components/score/score.yml")

我们可在函数中生成管道：

@pipeline()
def uci_heart_classifier_scorer(
    input_data: Input(type=AssetTypes.URI_FOLDER), score_mode: str
):
    """This pipeline demonstrates how to make batch inference using a model from the Heart Disease Data Set problem, where pre and post processing is required as steps. The pre and post processing steps can be components reusable from the training pipeline."""
    prepared_data = prepare_data(
        data=input_data,
        transformations=Input(type=AssetTypes.CUSTOM_MODEL, path=transformations.id),
    )
    scored_data = score_data(
        data=prepared_data.outputs.prepared_data,
        model=Input(type=AssetTypes.MLFLOW_MODEL, path=model.id),
        score_mode=score_mode,
    )

    return {"scores": scored_data.outputs.scores}

管道的可视化效果如下所示：

测试管道

我们可使用一些示例数据来测试管道。为此，我们将使用之前创建的管道和 batch-cluster 计算群集创建一个作业。

Azure CLI
Python

以下 pipeline-job.yml 文件包含了管道作业的配置：

pipeline-job.yml

$schema: https://azuremlschemas.azureedge.net/latest/pipelineJob.schema.json
type: pipeline

display_name: uci-classifier-score-job
description: |-
  This pipeline demonstrate how to make batch inference using a model from the Heart \
  Disease Data Set problem, where pre and post processing is required as steps. The \
  pre and post processing steps can be components reused from the training pipeline.

compute: batch-cluster
component: pipeline.yml
inputs:
  input_data:
    type: uri_folder
  score_mode: append
outputs: 
  scores:
    mode: upload

pipeline_job = uci_heart_classifier_scorer(
    input_data=Input(type="uri_folder", path="data/unlabeled/"), score_mode="append"
)

现在，我们将配置一些运行设置来运行测试：

pipeline_job.settings.default_datastore = "workspaceblobstore"
pipeline_job.settings.default_compute = "batch-cluster"

创建测试作业：

Azure CLI
Python

az ml job create -f pipeline-job.yml --set inputs.input_data.path=data/unlabeled

pipeline_job_run = ml_client.jobs.create_or_update(
    pipeline_job, experiment_name="uci-heart-score-pipeline"
)
pipeline_job_run

创建批处理终结点

为终结点提供名称。批处理终结点的名称在每个区域中必须是唯一的，因为该名称将用于构造调用 URI。为了确保唯一性，请在以下代码中指定的名称后面追加任何尾随字符。
- Azure CLI
- Python
```
ENDPOINT_NAME="uci-classifier-score"
```
```
endpoint_name = "uci-classifier-score"
```

配置终结点：

Azure CLI
Python

endpoint.yml 文件包含了终结点的配置。

endpoint.yml

$schema: https://azuremlschemas.azureedge.net/latest/batchEndpoint.schema.json
name: uci-classifier-score
description: Batch scoring endpoint of the Heart Disease Data Set prediction task.
auth_mode: aad_token

endpoint = BatchEndpoint(
    name=endpoint_name,
    description="Batch scoring endpoint of the Heart Disease Data Set prediction task",
)

创建终结点：

Azure CLI
Python

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

ml_client.batch_endpoints.begin_create_or_update(endpoint).result()

查询终结点 URI：

Azure CLI
Python

az ml batch-endpoint show --name $ENDPOINT_NAME

endpoint = ml_client.batch_endpoints.get(name=endpoint_name)
print(endpoint)

部署管道组件

要部署管道组件，必须创建批处理部署。部署是承载执行实际工作的资产所需的一组资源。

配置部署

Azure CLI
Python

deployment.yml 文件包含部署的配置。可以检查额外属性中的完整批处理终结点 YAML 机构。

deployment.yml

$schema: https://azuremlschemas.azureedge.net/latest/pipelineComponentBatchDeployment.schema.json
name: uci-classifier-prepros-xgb
endpoint_name: uci-classifier-batch
type: pipeline
component: pipeline.yml
settings:
    continue_on_step_failure: false
    default_compute: batch-cluster

管道是在函数中定义的。若要将其转换为组件，请从中使用 component 属性。管道组件是可重用的计算图，它们可以包含在批处理部署中，也可以用于组合更复杂的管道。

pipeline_component = ml_client.components.create_or_update(
    uci_heart_classifier_scorer().component
)

现在，我们可以定义部署：

deployment = PipelineComponentBatchDeployment(
    name="uci-classifier-prepros-xgb",
    description="A sample deployment with pre and post processing done before and after inference.",
    endpoint_name=endpoint.name,
    component=pipeline_component,
    settings={"continue_on_step_failure": False, "default_compute": compute_name},
)

创建部署
- Azure CLI
- Python
运行以下代码以在批处理终结点下创建一个批处理部署，并将其设置为默认部署。
```
az ml batch-deployment create --endpoint $ENDPOINT_NAME -f deployment.yml --set-default
```
提示

请注意，我们使用 --set-default 标志来指示此新部署现在是默认部署。
此命令将启动部署创建操作，并在部署创建操作继续时返回确认响应。
```
ml_client.batch_deployments.begin_create_or_update(deployment).result()
```
创建后，将此新部署配置为默认部署：
```
endpoint = ml_client.batch_endpoints.get(endpoint_name)
endpoint.defaults.deployment_name = deployment.name
ml_client.batch_endpoints.begin_create_or_update(endpoint).result()
```
部署已就绪，可供使用。

测试部署

创建部署后，即可接收作业。按照以下步骤对其进行测试：

部署要求指示一个数据输入和一个文本输入。

Azure CLI
Python

inputs.yml 文件包含了输入数据资产的定义：

inputs.yml

inputs:
  input_data:
    type: uri_folder
    path: data/unlabeled
  score_mode:
    type: string
    default: append
outputs:
  scores:
    type: uri_folder
    mode: upload

输入数据资产定义：

input_data = Input(type="uri_folder", path="data/unlabeled/")
score_mode = Input(type="string", default="append")

提示

要详细了解如何指示输入，请参阅《为批处理终结点创建作业和输入数据》。

可以按如下所示调用默认部署：
- Azure CLI
- Python
```
JOB_NAME=$(az ml batch-endpoint invoke -n $ENDPOINT_NAME --f inputs.yml --query name -o tsv)
```
提示

调用终结点时，使用 inputs 和 input 参数有什么差别？

一般情况下，可以将字典 inputs = {} 参数与 invoke 方法一起使用，以便向包含模型部署或管道部署的批处理终结点提供任意数量的所需输入。

对于模型部署，可以使用 参数来更方便地指定部署的输入数据位置input。这种方法之所以可行，是因为模型部署始终只采用一个数据输入。
```
job = ml_client.batch_endpoints.invoke(
    endpoint_name=endpoint.name,
    inputs={"input_data": input_data, "score_mode": score_mode},
)
```
可以使用以下方法监视演示进度并流式传输日志：
- Azure CLI
- Python
```
az ml job stream -n $JOB_NAME
```
```
ml_client.jobs.get(job.name)
```
要等待作业完成，请运行以下代码：
```
ml_client.jobs.stream(name=job.name)
```

访问作业输出

作业完成后，可以访问其输出。此作业仅包含一个名为 scores 的输出：

Azure CLI
Python

可以使用 az ml job download 下载关联的结果。

az ml job download --name $JOB_NAME --output-name scores

下载结果：

ml_client.jobs.download(name=job.name, download_path=".", output_name="scores")

读取评分数据：

import pandas as pd
import glob

output_files = glob.glob("named-outputs/scores/*.csv")
score = pd.concat((pd.read_csv(f) for f in output_files))
score

输出如下所示：

age	sex	...	thal	prediction
0.9338	1	...	2	0
1.3782	1	...	3	1
1.3782	1	...	4	0
-1.954	1	...	3	0

输出包含预测以及提供给已预处理的评分组件的数据。例如，列 age 已规范化，并且列 thal 包含原始编码值。实际上，你可能只想输出预测，然后将其与原始值相连接。此工作已留给读者完成。

清理资源

完成后，从工作区中删除关联的资源：

Azure CLI
Python

运行以下代码以删除批处理终结点及其基础部署。 --yes 用于确认删除。

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

删除终结点：

ml_client.batch_endpoints.begin_delete(endpoint_name)

（可选）除非计划在以后的部署中重用计算群集，否则请删除计算。

Azure CLI
Python

az ml compute delete -n batch-cluster

ml_client.compute.begin_delete(name="batch-cluster")

通过