如何透過前置處理部署管線，以執行批次評分

發行項
07/17/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 Machine Learning。
Azure Machine Learning 工作區。如果您沒有帳戶，請使用管理 Azure Machine Learning 工作區一文中的步驟來建立帳戶。
請確定您在工作區中具有下列權限：
- 建立或管理批次端點和部署：使用允許 Microsoft.MachineLearningServices/workspaces/batchEndpoints/* 的擁有者、參與者或自訂角色。
- 在工作區資源群組中建立 ARM 部署：在部署工作區的資源群組中，使用允許 Microsoft.Resources/deployments/write 的擁有者、參與者或自訂角色。
您必須安裝下列軟體，才能使用 Azure Machine Learning：
- Azure CLI
- Python
Azure CLI 和ml適用於 Azure Machine Learning 的延伸模組。
```
az extension add -n ml
```
注意

批次端點的管線元件部署是在 Azure CLI 的 ml 擴充功能 2.7 版中引進。使用 az extension update --name ml 來取得其最後一個版本。
適用於 Python 的 Azure Machine Learning SDK。
```
pip install azure-ai-ml
```
注意

類別 ModelBatchDeployment 和 PipelineComponentBatchDeployment 是在 SDK 1.7.0 版中引進。使用 pip install -U azure-ai-ml 來取得其最後一個版本。

連線到您的工作區

工作區是 Azure Machine Learning 的最上層資源，其提供一個集中位置來處理您在使用 Azure Machine Learning 時建立的所有成品。在本節中，我們將連線到您將執行部署工作的工作區。

Azure CLI
Python

在下列程式碼中傳入訂用帳戶識別碼、工作區、位置和資源群組的值：

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

設定工作區詳細資料，並取得工作區的控制代碼：

在下列程式碼中傳入訂用帳戶識別碼、工作區和資源群組的值：

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 Machine Learning 計算叢集上執行。因此，多個批次部署可以共用相同的計算基礎結構。在此範例中，我們將處理名為 batch-cluster 的 Azure Machine Learning 計算叢集。我們須確認計算存在於工作區上，否則要另外建立。

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

輸出如下所示：

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

輸出包含預測以及提供給 score 元件的數據，該元件已進行前置處理。舉例來說，資料行 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")

共用方式為