Come distribuire una pipeline per eseguire l'assegnazione dei punteggi in batch con 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 assegnazione dei punteggi) in un endpoint batch. La pipeline esegue l’assegnazione dei punteggi 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 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 accertarsi 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 dal 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 YAML e altri file, usare i comandi seguenti per clonare il repository e passare alla cartella per il linguaggio di codifica:

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 Jupyter Notebook

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

Prerequisiti

• Una sottoscrizione di Azure. Se non si ha una sottoscrizione di Azure, creare un account gratuito prima di iniziare.

• Un'area di lavoro di Azure Machine Learning. Per creare un'area di lavoro, vedere Gestire le aree di lavoro di Azure Machine Learning.

• Le autorizzazioni seguenti nell'area di lavoro di Azure Machine Learning:

  • Per la creazione o la gestione di endpoint e distribuzioni batch: usare un ruolo proprietario, collaboratore o personalizzato a cui sono state assegnate le Microsoft.MachineLearningServices/workspaces/batchEndpoints/* autorizzazioni.
  • Per la creazione di distribuzioni di Azure Resource Manager nel gruppo di risorse dell'area di lavoro: usare un ruolo proprietario, collaboratore o personalizzato a cui è stata assegnata l'autorizzazione Microsoft.Resources/deployments/write nel gruppo di risorse in cui viene distribuita l'area di lavoro.

• Interfaccia della riga di comando di Azure Machine Learning o Azure Machine Learning SDK per Python:

  Interfaccia della riga di comando di Azure

  Eseguire il comando seguente per installare l'interfaccia della riga di comando di Azure e l'mlestensione per Azure Machine Learning:

  az extension add -n ml
  

  Le distribuzioni dei componenti della pipeline per gli endpoint batch vengono introdotte nella versione 2.7 dell'estensione per l'interfaccia ml della riga di comando di Azure. Usare il comando az extension update --name ml per ottenere la versione più recente.

  Python

  Eseguire il comando seguente per installare Azure Machine Learning SDK per Python:

  pip install azure-ai-ml
  

  Le classi ModelBatchDeployment e PipelineComponentBatchDeployment vengono introdotte nella versione 1.7.0 dell'SDK. Usare il comando pip install -U azure-ai-ml per ottenere la versione più recente.

---

Connettersi all'area di lavoro

L'area di lavoro è la risorsa di primo livello per Azure Machine Learning. Offre una posizione centralizzata per lavorare con tutti gli artefatti creati quando si usa Azure Machine Learning. In questa sezione ci si connette all'area di lavoro in cui verranno eseguite le attività di distribuzione.

Interfaccia della riga di comando di Azure

Nel comando seguente immettere l'ID sottoscrizione, il nome dell'area di lavoro, il nome del gruppo di risorse e il percorso:

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:

   Nel comando seguente immettere l'ID sottoscrizione, il nome del gruppo di risorse e il nome dell'area di lavoro:

   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. Successivamente, verrà creato un cluster di elaborazione in cui verrà eseguita la distribuzione batch. Successivamente, verranno registrati i componenti, le trasformazioni e i modelli necessari per generare la pipeline di inferenza. Infine, la pipeline verrà generata e testata.

Creare l'ambiente

I componenti di questo esempio useranno un ambiente con le librerie XGBoost e scikit-learn. Il file environment/conda.yml 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 elaborazione. Possono essere eseguiti in qualsiasi cluster di elaborazione 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 utilizzerà un cluster di elaborazione di Azure Machine Learning denominato batch-cluster. Verificare che il calcolo esista nell'area di lavoro o, in caso contrario, crearlo.

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 i componenti, le trasformazioni e i modelli necessari per la generazione della 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 gli 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. I dati di input, invece, sono stati pre-elaborati (o trasformati) prima del training, usando un componente di preparazione. Sarà necessario registrare anche 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 la registrazione del componente di preparazione, è possibile farvi riferimento dall'area di lavoro. Ad esempio, azureml:uci_heart_prepare@latest otterrà l'ultima versione del componente di preparazione.

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

   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. Verrà eseguita 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 sua definizione per vedere quali componenti vengono riutilizzati dalla pipeline di training e quali sono nuovi.

Creare la pipeline

A questo punto, è possibile associare tutti gli elementi. 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: una cartella contenente i dati di input a cui assegnare i punteggi
  • transformations (facoltativo): il percorso delle trasformazioni che verranno applicate, se disponibile. Se è specificato, le trasformazioni vengono lette dal modello indicato nel percorso. Se il percorso non viene specificato, tuttavia, 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 preprocess_job può essere usato durante il training e l'assegnazione dei punteggi.
• score_job: questo passaggio eseguirà l'inferenza sui dati trasformati usando il modello di input. Tenere presente che il componente usa un modello MLflow per eseguire l'inferenza. Infine, i punteggi vengono riscritti nello stesso formato in cui sono stati letti.

Interfaccia della riga di comando di Azure

La configurazione della pipeline è definita nel file 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}}

Python

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

Generare 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 campione. A tale scopo, verrà creato un processo usando la pipeline e il cluster di elaborazione batch-cluster 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, configurare 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. Immettere 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 trasformarla in un componente, si userà la proprietà component della pipeline. 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
   )

   A questo punto è 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 che l'uso del flag --set-default sta ad indicare che ora questa nuova distribuzione è la distribuzione 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 testarla, effettuare la procedura seguente:

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

   Interfaccia della riga di comando di Azure

   Il file inputs.yml 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

   La definizione dell’asset dei 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 il parametro inputs e input quando si richiama un endpoint?

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

   Per una distribuzione modello, è possibile usare il parametro input come modo più breve per specificare la posizione dei dati di input per la distribuzione. Questo approccio funziona perché 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, usare 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 a cui sono stati assegnati i punteggi

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 punteggio pre-elaborato. Ad esempio, la colonna age è stata normalizzata e la colonna thal contiene valori di codifica originari. In pratica, è probabile che si desideri solo l’output della stima e si desideri concatenarlo 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 preveda di riutilizzare il cluster di elaborazione 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 dai processi della pipeline
• Accesso ai dati dai processi di endpoint batch
• Risoluzione dei problemi relativi agli endpoint batch