Come distribuire una pipeline per eseguire l'assegnazione dei punteggi batch con la pre-elaborazione

SI APPLICA A:Estensione ML dell'interfaccia della riga di comando di Azure v2 (corrente)Python SDK azure-ai-ml v2 (corrente)

Questo articolo illustra come distribuire una pipeline di inferenza (o punteggio) in un endpoint batch. La pipeline esegue il punteggio su un modello registrato, riutilizzando al tempo stesso un componente di pre-elaborazione da quando è stato eseguito il training del modello. Il riutilizzo dello stesso componente di pre-elaborazione garantisce che la stessa pre-elaborazione venga applicata durante l'assegnazione dei punteggi.

Si apprenderà come:

• Creare una pipeline che riutilizza i componenti esistenti dall'area di lavoro
• Distribuire la pipeline in un endpoint
• Usare le stime generate dalla pipeline

Informazioni su questo esempio

Questo esempio illustra come riutilizzare il codice di pre-elaborazione e i parametri appresi durante la pre-elaborazione prima di usare il modello per l'inferenza. Riutilizzando il codice di pre-elaborazione e i parametri appresi, è possibile assicurarsi che durante l'inferenza vengano applicate anche le stesse trasformazioni(ad esempio la normalizzazione e la codifica delle funzionalità) applicate ai dati di input durante il training. Il modello usato per l'inferenza eseguirà stime sui dati tabulari del set di dati UCI Heart Disease.

Una visualizzazione della pipeline è la seguente:

L'esempio contenuto in questo articolo si basa sugli esempi di codice contenuti nel repository azureml-examples. Per eseguire i comandi in locale senza dover copiare o incollare il file YAML e altri file, innanzitutto clonare il repository quindi cambiare le directory nella cartella:

Interfaccia della riga di comando di Azure

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

Python

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

I file per questo esempio si trovano in:

cd endpoints/batch/deploy-pipelines/batch-scoring-with-preprocessing

Seguire la procedura nei notebook di Jupyter

È possibile seguire la versione di Python SDK di questo esempio aprendo il notebook sdk-deploy-and-test.ipynb nel repository clonato.

Prerequisiti

Prima di seguire la procedura descritta in questo articolo, assicurarsi di disporre dei prerequisiti seguenti:

• Una sottoscrizione di Azure. Se non si ha una sottoscrizione di Azure, creare un account gratuito prima di iniziare. Provare la versione gratuita o a pagamento di Azure Machine Learning.

• Un'area di lavoro di Azure Machine Learning. Se non è disponibile, usare la procedura descritta nell'articolo Gestire le aree di lavoro di Azure Machine Learning per crearne una.

• Assicurarsi di disporre delle autorizzazioni seguenti nell'area di lavoro:

  • Creare o gestire endpoint e distribuzioni batch: usare un ruolo Proprietario, Collaboratore o Personalizzato che consenta Microsoft.MachineLearningServices/workspaces/batchEndpoints/*.

  • Creare distribuzioni arm nel gruppo di risorse dell'area di lavoro: usare un ruolo Proprietario, Collaboratore o Personalizzato che consenta Microsoft.Resources/deployments/write nel gruppo di risorse in cui viene distribuita l'area di lavoro.

• Per usare Azure Machine Learning, è necessario installare il software seguente:

  Interfaccia della riga di comando di Azure

  L'Interfaccia della riga di comando di Azure e l'mlestensione per Azure Machine Learning.

  az extension add -n ml
  

  Nota

  Le distribuzioni dei componenti della pipeline per gli endpoint batch sono state introdotte nella versione 2.7 dell'estensione ml per l'interfaccia della riga di comando di Azure. Usare az extension update --name ml per ottenere l'ultima versione.

  Python

  SDK Azure Machine Learning per Python.

  pip install azure-ai-ml
  

  Nota

  Le classi ModelBatchDeployment e PipelineComponentBatchDeployment sono state introdotte nella versione 1.7.0 dell'SDK. Usare pip install -U azure-ai-ml per ottenere l'ultima versione.

Connettersi all'area di lavoro

L'area di lavoro è la risorsa di primo livello per Azure Machine Learning, che fornisce una posizione centralizzata da cui gestire tutti gli artefatti creati quando si usa Azure Machine Learning. In questa sezione ci si connetterà all'area di lavoro in cui verranno eseguite le attività di distribuzione.

Interfaccia della riga di comando di Azure

Passare i valori per l'ID sottoscrizione, l'area di lavoro, la posizione e il gruppo di risorse nel codice seguente:

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

Python

1. Importare le librerie necessarie:

   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
   

2. Configurare i dettagli dell'area di lavoro e ottenere un handle per l'area di lavoro:

   Passare i valori per l'ID sottoscrizione, l'area di lavoro e il gruppo di risorse nel codice seguente:

   subscription_id = "<subscription>"
   resource_group = "<resource-group>"
   workspace = "<workspace>"
   
   ml_client = MLClient(DefaultAzureCredential(), subscription_id, resource_group, workspace)
   

Creare la pipeline di inferenza

In questa sezione verranno creati tutti gli asset necessari per la pipeline di inferenza. Si inizierà creando un ambiente che include le librerie necessarie per i componenti della pipeline. Verrà quindi creato un cluster di calcolo in cui verrà eseguita la distribuzione batch. Successivamente, verranno registrati i componenti, i modelli e le trasformazioni necessari per compilare la pipeline di inferenza. Infine, verrà compilata e testata la pipeline.

Creare l'ambiente

I componenti di questo esempio useranno un ambiente con le XGBoost librerie e scikit-learn . Il environment/conda.yml file contiene la configurazione dell'ambiente:

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

Creare l'ambiente nel modo seguente:

1. Definire l'ambiente:

   Interfaccia della riga di comando di Azure

   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.
   

   Python

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

2. Creare l'ambiente:

   Interfaccia della riga di comando di Azure

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

   Python

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

Creare un cluster di elaborazione

Gli endpoint e le distribuzioni batch vengono eseguiti nei cluster di calcolo. Possono essere eseguiti in qualsiasi cluster di calcolo di Azure Machine Learning già esistente nell'area di lavoro. Di conseguenza, più distribuzioni batch possono condividere la stessa infrastruttura di calcolo. In questo esempio si lavorerà in un cluster di calcolo di Azure Machine Learning denominato batch-cluster. Verificare che il calcolo esista nell'area di lavoro o crearlo in caso contrario.

Interfaccia della riga di comando di Azure

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

Python

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()

Registrare componenti e modelli

Verranno registrati componenti, modelli e trasformazioni necessari per creare la pipeline di inferenza. È possibile riutilizzare alcuni di questi asset per le routine di training.

Suggerimento

In questa esercitazione verrà riutilizzato il modello e il componente di pre-elaborazione da una pipeline di training precedente. È possibile vedere come sono stati creati seguendo l'esempio Come distribuire una pipeline di training con endpoint batch.

1. Registrare il modello da usare per la stima:

   Interfaccia della riga di comando di Azure

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

   Python

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

2. Il modello registrato non è stato sottoposto a training direttamente sui dati di input. Al contrario, i dati di input sono stati pre-elaborati (o trasformati) prima del training, usando un componente di preparazione. È anche necessario registrare questo componente. Registrare il componente di preparazione:

   Interfaccia della riga di comando di Azure

   az ml component create -f components/prepare/prepare.yml
   

   Python

   prepare_data = load_component(source="components/prepare/prepare.yml")
   
   ml_client.components.create_or_update(prepare_data)

   Suggerimento

   Dopo aver registrato il componente di preparazione, è ora possibile farvi riferimento dall'area di lavoro. Ad esempio, azureml:uci_heart_prepare@latest otterrà l'ultima versione del componente di preparazione.

3. Nell'ambito delle trasformazioni dei dati nel componente di preparazione, i dati di input sono stati normalizzati per centrare i predittori e limitarne i valori nell'intervallo di [-1, 1]. I parametri di trasformazione sono stati acquisiti in una trasformazione scikit-learn che è anche possibile registrare per applicarli in un secondo momento quando sono disponibili nuovi dati. Registrare la trasformazione come segue:

   Interfaccia della riga di comando di Azure

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

   Python

   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,
       )
   )

4. Si eseguirà l'inferenza per il modello registrato, usando un altro componente denominato score che calcola le stime per un determinato modello. Si farà riferimento al componente direttamente dalla relativa definizione.

   Suggerimento

   La procedura consigliata consiste nel registrare il componente e farvi riferimento dalla pipeline. In questo esempio, tuttavia, si farà riferimento al componente direttamente dalla relativa definizione per vedere quali componenti vengono riutilizzati dalla pipeline di training e quali sono nuovi.

Creare la pipeline

Ora è il momento di associare tutti gli elementi insieme. La pipeline di inferenza che verrà distribuita include due componenti (passaggi):

• preprocess_job: questo passaggio legge i dati di input e restituisce i dati preparati e le trasformazioni applicate. Il passaggio riceve due input:
  • data: cartella contenente i dati di input da assegnare al punteggio
  • transformations: (facoltativo) Percorso delle trasformazioni che verranno applicate, se disponibili. Se specificato, le trasformazioni vengono lette dal modello indicato nel percorso. Tuttavia, se il percorso non viene specificato, le trasformazioni verranno apprese dai dati di input. Per l'inferenza, tuttavia, non è possibile apprendere i parametri di trasformazione (in questo esempio, i coefficienti di normalizzazione) dai dati di input perché è necessario usare gli stessi valori dei parametri appresi durante il training. Poiché questo input è facoltativo, il componente può essere usato durante il preprocess_job training e l'assegnazione dei punteggi.
• score_job: questo passaggio eseguirà l'inferenza sui dati trasformati usando il modello di input. Si noti che il componente usa un modello MLflow per eseguire l'inferenza. Infine, i punteggi vengono riscritto nello stesso formato in cui sono stati letti.

Interfaccia della riga di comando di Azure

La configurazione della pipeline.yml pipeline è definita nel file :

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}}

Python

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

Compilare la pipeline in una funzione:

@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}

Una visualizzazione della pipeline è la seguente:

Testare la pipeline

Verrà ora testata la pipeline con alcuni dati di esempio. A tale scopo, verrà creato un processo usando la pipeline e il batch-cluster cluster di calcolo creato in precedenza.

Interfaccia della riga di comando di Azure

Il file pipeline-job.yml seguente contiene la configurazione per il processo della pipeline:

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

Python

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

A questo punto, verranno configurate alcune impostazioni di esecuzione per eseguire il test:

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

Creare il processo di test:

Interfaccia della riga di comando di Azure

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

Python

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

Creare un endpoint batch

1. Specificare un nome per l'endpoint. Il nome di un endpoint batch deve essere univoco in ogni area poiché viene usato per costruire l'URI di chiamata. Per garantirne l'univocità, aggiungere tutti i caratteri finali al nome specificato nel codice seguente.

   Interfaccia della riga di comando di Azure

   ENDPOINT_NAME="uci-classifier-score"
   

   Python

   endpoint_name = "uci-classifier-score"

2. Configurare l'endpoint:

   Interfaccia della riga di comando di Azure

   Il file endpoint.yml contiene la configurazione dell'endpoint.

   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
   

   Python

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

3. Creare l'endpoint:

   Interfaccia della riga di comando di Azure

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

   Python

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

4. Eseguire una query sull'URI dell'endpoint:

   Interfaccia della riga di comando di Azure

   az ml batch-endpoint show --name $ENDPOINT_NAME
   

   Python

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

Distribuire il componente della pipeline

Per distribuire il componente della pipeline, è necessario creare una distribuzione batch. Una distribuzione è un set di risorse necessarie per ospitare l'asset che esegue il lavoro effettivo.

1. Configurare la distribuzione

   Interfaccia della riga di comando di Azure

   Il file deployment.yml contiene la configurazione della distribuzione. È possibile controllare lo schema YAML dell'endpoint batch completo per ottenere proprietà aggiuntive.

   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
   

   Python

   La pipeline è definita in una funzione. Per trasformarlo in un componente, si userà la component proprietà da essa. I componenti della pipeline sono grafici di calcolo riutilizzabili che possono essere inclusi nelle distribuzioni batch o usati per comporre pipeline più complesse.

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

   È ora possibile definire la distribuzione:

   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},
   )

2. Creare la distribuzione

   Interfaccia della riga di comando di Azure

   Eseguire il codice seguente per creare una distribuzione batch nell'endpoint batch e impostarla come distribuzione predefinita.

   az ml batch-deployment create --endpoint $ENDPOINT_NAME -f deployment.yml --set-default
   

   Suggerimento

   Si noti l'uso del --set-default flag per indicare che questa nuova distribuzione è ora l'impostazione predefinita.

   Python

   Questo comando avvia la creazione della distribuzione e restituisce una risposta di conferma mentre la procedura è ancora in corso.

   ml_client.batch_deployments.begin_create_or_update(deployment).result()

   Dopo la creazione, configurare questa nuova distribuzione come predefinita:

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

3. La distribuzione è pronta per l'uso.

Testare la distribuzione

Dopo aver creato la distribuzione, questa è pronta per ricevere i processi. Per testarlo, seguire questa procedura:

1. Per la distribuzione è necessario indicare un input di dati e un input letterale.

   Interfaccia della riga di comando di Azure

   Il inputs.yml file contiene la definizione per l'asset di dati di input:

   inputs.yml

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

   Python

   Definizione dell'asset di dati di input:

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

   Suggerimento

   Per altre informazioni su come indicare gli input, vedere Creare processi e dati di input per gli endpoint batch.

2. È possibile richiamare la distribuzione predefinita nel modo seguente:

   Interfaccia della riga di comando di Azure

   JOB_NAME=$(az ml batch-endpoint invoke -n $ENDPOINT_NAME --f inputs.yml --query name -o tsv)
   

   Python

   Suggerimento

   Qual è la differenza tra inputs e input quando si richiama un endpoint?

   In generale, è possibile usare un dizionario con il invoke metodo per fornire un numero arbitrario inputs = {} di input necessari a un endpoint batch che contiene una distribuzione del modello o una distribuzione della pipeline.

   Per una distribuzione del modello, è possibile usare input come modo più breve per specificare il percorso dei dati di input per la distribuzione, poiché una distribuzione del modello accetta sempre un solo input di dati.

   job = ml_client.batch_endpoints.invoke(
       endpoint_name=endpoint.name,
       inputs={"input_data": input_data, "score_mode": score_mode},
   )

3. È possibile monitorare lo stato di avanzamento del contenuto e trasmettere i log usando:

   Interfaccia della riga di comando di Azure

   az ml job stream -n $JOB_NAME
   

   Python

   ml_client.jobs.get(job.name)

   Per attendere il completamento del processo, eseguire il codice seguente:

   ml_client.jobs.stream(name=job.name)

Accedere all'output del processo

Una volta completato il processo, è possibile accedere al relativo output. Questo processo contiene un solo output denominato scores:

Interfaccia della riga di comando di Azure

È possibile scaricare i risultati associati usando az ml job download.

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

Python

Scaricare il risultato:

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

Leggere i dati con punteggio:

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

L'output sarà simile al seguente:

età	sex	...	thal	stima
0.9338	1	...	2	0
1.3782	1	...	3	1
1.3782	1	...	4	0
-1.954	1	...	3	0

L'output contiene le stime più i dati forniti al componente score, che è stato pre-elaborato. Ad esempio, la colonna age è stata normalizzata e la colonna thal contiene valori di codifica originali. In pratica, è probabile che si voglia restituire solo la stima e quindi concatarla con i valori originali. Questo lavoro è stato lasciato al lettore.

Pulire le risorse

Al termine, eliminare le risorse associate dall'area di lavoro:

Interfaccia della riga di comando di Azure

Eseguire il codice seguente per eliminare l'endpoint batch e la relativa distribuzione sottostante. --yes viene usato per confermare l'eliminazione.

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

Python

Eliminare l'endpoint:

ml_client.batch_endpoints.begin_delete(endpoint_name)

(Facoltativo) Eliminare le risorse di calcolo, a meno che non si intenda riutilizzare il cluster di calcolo con distribuzioni successive.

Interfaccia della riga di comando di Azure

az ml compute delete -n batch-cluster

Python

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

Passaggi successivi

• Creare endpoint batch da processi della pipeline
• Accesso ai dati dai processi di endpoint batch
• Risoluzione dei problemi relativi agli endpoint batch