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

使用批处理模型部署进行图像处理

项目
12/26/2023

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

批处理模型部署可用于处理表格数据，也可用于处理图像等任何其他文件类型。 MLflow 和自定义模型都支持这些部署。本教程介绍如何根据 ImageNet 分类部署图像分类模型。

关于此示例

我们将采用的模型使用 TensorFlow 和 RestNet 体系结构 (深度残差网络中的标识映射) 建成。可从此处下载此模型的示例。此模型具有以下约束，这些约束对于部署而言非常重要，请务必牢记：

适用于大小为 244x244 的图像（张量 (224, 224, 3)）。
要求将输入缩放到范围 [0,1] 内。

本文中的信息基于 azureml-examples 存储库中包含的代码示例。要在本地运行命令而不必复制/粘贴 YAML 和其他文件，请克隆存储库repo，然后将目录更改为 cli/endpoints/batch/deploy-models/imagenet-classifier（如果你使用的是 Azure CLI）或更改为 sdk/python/endpoints/batch/deploy-models/imagenet-classifier（如果你使用的是我们的 Python SDK）。

git clone https://github.com/Azure/azureml-examples --depth 1
cd azureml-examples/cli/endpoints/batch/deploy-models/imagenet-classifier

在 Jupyter Notebook 中继续操作

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

先决条件

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

Azure 订阅。如果没有 Azure 订阅，请在开始操作前先创建一个免费帐户。试用免费版或付费版 Azure 机器学习。
一个 Azure 机器学习工作区。如果你没有，请按照管理 Azure 机器学习工作区一文中的步骤创建一个。
确保你在工作区中具有以下权限：
- 创建或管理批量终结点和部署：使用允许 Microsoft.MachineLearningServices/workspaces/batchEndpoints/* 的所有者、参与者或自定义角色。
- 在工作区资源组中创建 ARM 部署：在部署工作区的资源组中使用允许 Microsoft.Resources/deployments/write 的所有者角色、参与者角色或自定义角色。
需要安装以下软件才能使用 Azure 机器学习：
- Azure CLI
- Python
Azure CLI 和ml适用于 Azure 机器学习的扩展。
```
az extension add -n ml
```
注意

Azure CLI 的 ml 扩展版本 2.7 中引入了 Batch 终结点的管道组件部署。使用 az extension update --name ml 获取它的上一个版本。
适用于 Python 的 Azure 机器学习 SDK。
```
pip install azure-ai-ml
```
注意

ModelBatchDeployment 和 PipelineComponentBatchDeployment 类是在 SDK 版本 1.7.0 中推出的。使用 pip install -U azure-ai-ml 获取它的上一个版本。

连接到工作区

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

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)

利用批部署进行图像分类

在此示例中，我们将了解如何部署可根据 ImageNet 分类对给定图像进行分类的深度学习模型。

创建终结点

首先，让我们创建将托管模型的终结点：

Azure CLI
Python

确定终结点的名称：

ENDPOINT_NAME="imagenet-classifier-batch"

以下 YAML 文件定义批处理终结点：

endpoint.yml

$schema: https://azuremlschemas.azureedge.net/latest/batchEndpoint.schema.json
name: imagenet-classifier-batch
description: A batch endpoint for performing image classification using a TFHub model ImageNet model.
auth_mode: aad_token

运行以下代码以创建终结点。

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

确定终结点的名称：

endpoint_name="imagenet-classifier-batch"

配置终结点：

endpoint = BatchEndpoint(
    name=endpoint_name,
    description="An batch service to perform ImageNet image classification",
)

运行以下代码以创建终结点：

ml_client.batch_endpoints.begin_create_or_update(endpoint)

注册模型

模型部署只能部署已注册的模型，因此我们需要先注册。如果打算部署的模型已注册，则可以跳过此步骤。

下载模型副本：

Azure CLI
Python

wget https://azuremlexampledata.blob.core.windows.net/data/imagenet/model.zip
unzip model.zip -d .

import os
import urllib.request
from zipfile import ZipFile

response = urllib.request.urlretrieve('https://azuremlexampledata.blob.core.windows.net/data/imagenet/model.zip', 'model.zip')

os.mkdirs("imagenet-classifier", exits_ok=True)
with ZipFile(response[0], 'r') as zip:
  model_path = zip.extractall(path="imagenet-classifier")

注册模型：

Azure CLI
Python

MODEL_NAME='imagenet-classifier'
az ml model create --name $MODEL_NAME --path "model"

model_name = 'imagenet-classifier'
model = ml_client.models.create_or_update(
    Model(name=model_name, path=model_path, type=AssetTypes.CUSTOM_MODEL)
)

创建评分脚本

我们需要创建评分脚本，该脚本应能读取批处理部署提供的映像并返回模型的分数。以下脚本：

表示init使用tensorflow中模块keras加载模型的函数。
表示run为批处理部署提供的每个小型批处理执行的函数。
函数run一次读取文件的一个图像
该run方法可将图像大小调整为模型的预期大小。
该run方法将图像重新缩放到范围[0,1]域，即模型的预期内容。
它会返回与预测有关的类和概率。

code/score-by-file/batch_driver.py

import os
import numpy as np
import pandas as pd
import tensorflow as tf
from os.path import basename
from PIL import Image
from tensorflow.keras.models import load_model


def init():
    global model
    global input_width
    global input_height

    # AZUREML_MODEL_DIR is an environment variable created during deployment
    model_path = os.path.join(os.environ["AZUREML_MODEL_DIR"], "model")

    # load the model
    model = load_model(model_path)
    input_width = 244
    input_height = 244


def run(mini_batch):
    results = []

    for image in mini_batch:
        data = Image.open(image).resize(
            (input_width, input_height)
        )  # Read and resize the image
        data = np.array(data) / 255.0  # Normalize
        data_batch = tf.expand_dims(
            data, axis=0
        )  # create a batch of size (1, 244, 244, 3)

        # perform inference
        pred = model.predict(data_batch)

        # Compute probabilities, classes and labels
        pred_prob = tf.math.reduce_max(tf.math.softmax(pred, axis=-1)).numpy()
        pred_class = tf.math.argmax(pred, axis=-1).numpy()

        results.append([basename(image), pred_class[0], pred_prob])

    return pd.DataFrame(results)

提示

尽管图像由部署在小型批处理中提供，但此评分脚本一次仅处理一个图像。这是一种常见模式，尝试加载整个批处理并将其一次性发送到模型可能会导致批处理执行程序（OOM 执行程序）的内存压力增大。但是，在某些情况下，这样做会在评分任务中出现高吞吐量。例如，通过 GPU 硬件进行批量部署，我们想实现高 GPU 利用率时。有关基于它的评分脚本示例，请参阅高吞吐量部署。

注意

如想部署生成模型 (生成文件的模型) ，请参阅产生多个文件的模型部署中的如何编写评分脚本。

创建部署

评分脚本创建完成后，开始创建批处理部署。按照以下步骤进行创建：

确保创建了一个可在其中创建部署的计算群集。本示例将使用名为 gpu-cluster 的计算群集。尽管这不是必需的，但我们使用 GPU 来加快处理速度。

我们需要指出我们将在哪个环境中运行部署。在本例中，模型在 TensorFlow 上运行。 Azure 机器学习已具备已安装所需软件的环境，我们可以使用此环境。只需在conda.yml文件中添加几个依赖项即可。

Azure CLI
Python

环境定义将包含在部署文件中。

compute: azureml:gpu-cluster
environment:
  name: tensorflow212-cuda11-gpu
  image: mcr.microsoft.com/azureml/curated/tensorflow-2.12-cuda11:latest

获取对环境的引用：

environment = Environment(
    name="tensorflow27-cuda11-gpu",
    conda_file="environment/conda.yml",
    image="mcr.microsoft.com/azureml/curated/tensorflow-2.7-ubuntu20.04-py38-cuda11-gpu:latest",
)

现在，创建部署。

Azure CLI
Python

如需在创建的终结点下创建新部署，请创建如下所示的 YAML 配置。可以检查额外属性中的完整批处理终结点 YAML 机构。

$schema: https://azuremlschemas.azureedge.net/latest/modelBatchDeployment.schema.json
endpoint_name: imagenet-classifier-batch
name: imagenet-classifier-resnetv2
description: A ResNetV2 model architecture for performing ImageNet classification in batch
type: model
model: azureml:imagenet-classifier@latest
compute: azureml:gpu-cluster
environment:
  name: tensorflow212-cuda11-gpu
  image: mcr.microsoft.com/azureml/curated/tensorflow-2.12-cuda11:latest
  conda_file: environment/conda.yaml
code_configuration:
  code: code/score-by-file
  scoring_script: batch_driver.py
resources:
  instance_count: 2
settings:
  max_concurrency_per_instance: 1
  mini_batch_size: 5
  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-by-file.yml --endpoint-name $ENDPOINT_NAME --set-default

如需用所示环境和评分脚本创建新部署，请使用以下代码：

deployment = BatchDeployment(
    name="imagenet-classifier-resnetv2",
    description="A ResNetV2 model architecture for performing ImageNet classification in batch",
    endpoint_name=endpoint.name,
    model=model,
    environment=environment,
    code_configuration=CodeConfiguration(
        code="code/score-by-file",
        scoring_script="batch_driver.py",
    ),
    compute=compute_name,
    instance_count=2,
    max_concurrency_per_instance=1,
    mini_batch_size=10,
    output_action=BatchDeploymentOutputAction.APPEND_ROW,
    output_file_name="predictions.csv",
    retry_settings=BatchRetrySettings(max_retries=3, timeout=300),
    logging_level="info",
)

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

ml_client.batch_deployments.begin_create_or_update(deployment)

尽管可以在终结点内调用特定部署，但通常需要调用终结点本身，让终结点决定使用哪个部署。此类部署命名为“默认”部署。这使你可以更改默认部署，从而更改为部署提供服务的模型，而无需更改用户的调用终结点的协定。使用以下说明更新默认部署：
- Azure 机器学习 CLI
- 适用于 Python 的 Azure 机器学习 SDK
```
az ml batch-endpoint update --name $ENDPOINT_NAME --set defaults.deployment_name=$DEPLOYMENT_NAME
```
```
endpoint.defaults.deployment_name = deployment.name
ml_client.batch_endpoints.begin_create_or_update(endpoint)
```
此时，我们的批处理终结点随时可用。

测试部署

为测试终结点，我们将使用原始 ImageNet 数据集中的 1000 个图像示例。批处理终结点只能处理云中可从 Azure 机器学习工作区访问的数据。在该示例中，我们会将其上传到 Azure 机器学习数据存储中。具体地说，是我们将创建可用于调用终结点进行评分的数据资产。但是，请注意，批处理终结点接受可存放在多个位置的数据。

下载相关的示例数据：

Azure CLI
Python

wget https://azuremlexampledata.blob.core.windows.net/data/imagenet/imagenet-1000.zip
unzip imagenet-1000.zip -d data

!wget https://azuremlexampledata.blob.core.windows.net/data/imagenet-1000.zip
!unzip imagenet-1000.zip -d data

现在，我们用刚刚下载的数据创建数据资产

Azure CLI
Python

在YAML中创建数据资产定义：

imagenet-sample-unlabeled.yml

$schema: https://azuremlschemas.azureedge.net/latest/data.schema.json
name: imagenet-sample-unlabeled
description: A sample of 1000 images from the original ImageNet dataset. Download content from https://azuremlexampledata.blob.core.windows.net/data/imagenet-1000.zip.
type: uri_folder
path: data

然后，创建数据资产：

az ml data create -f imagenet-sample-unlabeled.yml

data_path = "data"
dataset_name = "imagenet-sample-unlabeled"

imagenet_sample = Data(
    path=data_path,
    type=AssetTypes.URI_FOLDER,
    description="A sample of 1000 images from the original ImageNet dataset",
    name=dataset_name,
)

然后，创建数据资产：

ml_client.data.create_or_update(imagenet_sample)

若要获取新创建的数据资产，请使用：

imagenet_sample = ml_client.data.get(dataset_name, label="latest")

数据已上传且随时可用，接下来我们将调用终结点：
- Azure CLI
- Python
```
JOB_NAME=$(az ml batch-endpoint invoke --name $ENDPOINT_NAME --input azureml:imagenet-sample-unlabeled@latest --query name -o tsv)
```
注意

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

调用终结点时，inputs 与 input 有何区别？

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

对于模型部署，可以使用 input 来更简洁地指定部署的输入数据位置，因为模型部署始终只使用一个数据输入。
```
input = Input(type=AssetTypes.URI_FOLDER, path=imagenet_sample.id)
job = ml_client.batch_endpoints.invoke(
   endpoint_name=endpoint.name,
   input=input,
)
```
提示

请注意，我们不会在调用操作中给出部署名称，因为终结点会自动将作业路由到默认部署。由于终结点只有一个部署，因此该部署为默认部署。可通过指明自变量/参数 deployment_name 来将特定部署作为目标。
命令返回后立即启动批处理作业。在作业完成前可监视作业状态：
- Azure CLI
- Python
```
az ml job show -n $JOB_NAME --web
```
```
ml_client.jobs.get(job.name)
```

部署完成后，可下载预测：

Azure CLI
Python

如需下载预测，请使用以下命令：

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

ml_client.jobs.download(name=job.name, output_name='score', download_path='./')

输出预测如下所示。请注意，为方便读者阅读，预测已与标签合并。如需有关如何操作的详细信息，请参阅相关笔记本。

import pandas as pd
score = pd.read_csv("named-outputs/score/predictions.csv", header=None,  names=['file', 'class', 'probabilities'], sep=' ')
score['label'] = score['class'].apply(lambda pred: imagenet_labels[pred])
score

文件	class	概率	label
n02088094_Afghan_hound.JPEG	161	0.994745	阿富汗猎犬
n02088238_basset	162	0.999397	basset
n02088364_beagle.JPEG	165	0.366914	bluetick
n02088466_bloodhound.JPEG	164	0.926464	侦探猎犬
...	...	...	...

高吞吐量部署

如前所述，我们刚刚创建的部署一次可处理一张图像，即使批处理部署提供了一批图像也是如此。多数情况下，这都是最佳方法，因为它简化了模型的执行方式，同时避免了任何可能出现的内存不足问题。但是，在某些情况下，我们可能希望尽量使用基础硬件。例如，GPU。

在这些情况下，我们可能需要对整批数据执行推理。这需要将整个图像集加载到内存中，并将其直接发送到模型。以下示例用TensorFlow读取批量图像并对全部图像一起评分。它还使用TensorFlow操作执行任何数据预处理，因此整个管道将出现在所使用 (CPU/GPU) 的同一设备上。

警告

某些模型与输入的大小在内存消耗方面存在非线性关系。再次批处理 (如本示例中所示) 或减小批处理部署创建的批大小，以免出现内存不足的异常情况。

创建评分脚本：

code/score-by-batch/batch_driver.py

import os
import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow.keras.models import load_model


def init():
    global model
    global input_width
    global input_height

    # AZUREML_MODEL_DIR is an environment variable created during deployment
    model_path = os.path.join(os.environ["AZUREML_MODEL_DIR"], "model")

    # load the model
    model = load_model(model_path)
    input_width = 244
    input_height = 244


def decode_img(file_path):
    file = tf.io.read_file(file_path)
    img = tf.io.decode_jpeg(file, channels=3)
    img = tf.image.resize(img, [input_width, input_height])
    return img / 255.0


def run(mini_batch):
    images_ds = tf.data.Dataset.from_tensor_slices(mini_batch)
    images_ds = images_ds.map(decode_img).batch(64)

    # perform inference
    pred = model.predict(images_ds)

    # Compute probabilities, classes and labels
    pred_prob = tf.math.reduce_max(tf.math.softmax(pred, axis=-1)).numpy()
    pred_class = tf.math.argmax(pred, axis=-1).numpy()

    return pd.DataFrame(
        [mini_batch, pred_prob, pred_class], columns=["file", "probability", "class"]
    )

提示

请注意，该脚本正在用批处理部署发送的小型批处理构建张量数据集。该数据集经预处理后可使用map函数运算decode_img获取模型的预期张量。
数据集再次批处理 (16)，将数据发送到模型。可使用该参数控制可以加载到内存中并一次发送到模型的信息量。如在 GPU 上运行，则需要对该参数进行调整，以在收到 OOM 异常的通知之前达到 GPU 的最大利用率。
预测计算完成后，张量将转换为numpy.ndarray。

现在，创建部署。

Azure CLI
Python

如需在创建的终结点下创建新部署，请创建如下所示的 YAML 配置。可以检查额外属性中的完整批处理终结点 YAML 机构。

$schema: https://azuremlschemas.azureedge.net/latest/modelBatchDeployment.schema.json
endpoint_name: imagenet-classifier-batch
name: imagenet-classifier-resnetv2
description: A ResNetV2 model architecture for performing ImageNet classification in batch
type: model
model: azureml:imagenet-classifier@latest
compute: azureml:gpu-cluster
environment:
  name: tensorflow212-cuda11-gpu
  image: mcr.microsoft.com/azureml/curated/tensorflow-2.12-cuda11:latest
  conda_file: environment/conda.yaml
code_configuration:
  code: code/score-by-batch
  scoring_script: batch_driver.py
resources:
  instance_count: 2
tags:
  device_acceleration: CUDA
  device_batching: 16
settings:
  max_concurrency_per_instance: 1
  mini_batch_size: 5
  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-by-batch.yml --endpoint-name $ENDPOINT_NAME --set-default

如需用所示环境和评分脚本创建新部署，请使用以下代码：

deployment = BatchDeployment(
    name="imagenet-classifier-resnetv2",
    description="A ResNetV2 model architecture for performing ImageNet classification in batch",
    endpoint_name=endpoint.name,
    model=model,
    environment=environment,
    code_configuration=CodeConfiguration(
        code="code/score-by-batch",
        scoring_script="batch_driver.py",
    ),
    compute=compute_name,
    instance_count=2,
    tags={ "device_acceleration": "CUDA", "device_batching": "16" }
    max_concurrency_per_instance=1,
    mini_batch_size=10,
    output_action=BatchDeploymentOutputAction.APPEND_ROW,
    output_file_name="predictions.csv",
    retry_settings=BatchRetrySettings(max_retries=3, timeout=300),
    logging_level="info",
)

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

ml_client.batch_deployments.begin_create_or_update(deployment)

可以将此新部署与前面显示的示例数据一起使用。请记住，若要调用此部署，应在调用方法中指明部署的名称，或将其设置为默认部署。

处理图像的 MLflow 模型的注意事项

批处理终结点中的 MLflow 模型支持将图像读取为输入数据。由于 MLflow 部署不需要评分脚本，因此使用它们时需要注意以下事项：

支持的图像文件包括：.png、.jpg、.jpeg、.tiff、.bmp和.gif。
MLflow 模型可能会收到与输入图像的尺寸相匹配的输入np.ndarray。为确保每次批处理均支持多种图像大小，批处理执行程序每个图像文件将调用一次 MLflow 模型。
强烈建议 MLflow 模型包含签名，若包含，则必须为TensorSpec类型。输入将重新调整为匹配张量的形状（如可用）。如果没有可用的签名，则推断类型为 np.uint8 的张量。
对于包含签名且能处理不同大小的图像的模型，应包含可以提供保证的签名。例如，以下签名示例将允许对 3 通道图像进行批处理。

import numpy as np
import mlflow
from mlflow.models.signature import ModelSignature
from mlflow.types.schema import Schema, TensorSpec

input_schema = Schema([
  TensorSpec(np.dtype(np.uint8), (-1, -1, -1, 3)),
])
signature = ModelSignature(inputs=input_schema)

(...)

mlflow.<flavor>.log_model(..., signature=signature)

可以在 Jupyter 笔记本 imagenet-classifier-mlflow.ipynb 中找到一个工作示例。有关如何在批处理部署中使用 MLflow 模型的详细信息，请参阅在批处理部署中使用 MLflow 模型。

Share via

使用批处理模型部署进行图像处理

关于此示例

在 Jupyter Notebook 中继续操作

先决条件

连接到工作区

利用批部署进行图像分类

创建终结点

注册模型

创建评分脚本

创建部署

测试部署

高吞吐量部署

处理图像的 MLflow 模型的注意事项

后续步骤

其他资源