Cómo implementar una canalización para realizar puntuaciones por lotes con preprocesamiento

Artículo
09/03/2024

SE APLICA A:Extensión ML de la CLI de Azure v2 (actual)SDK de Python azure-ai-ml v2 (actual)

En este artículo, aprenderá a implementar una canalización de inferencia (o puntuación) en un punto de conexión por lotes. La canalización realiza la puntuación sobre un modelo registrado mientras también reutiliza un componente de preprocesamiento desde el momento en que se entrenó el modelo. La reutilización del mismo componente de preprocesamiento garantizará que se aplique el mismo preprocesamiento durante la puntuación.

Aprenderá lo siguiente:

Creación de una canalización que reutilice los componentes existentes del área de trabajo
Implementación de la canalización en un punto de conexión
Consumo de predicciones generadas por la canalización

Acerca de este ejemplo

En este ejemplo, se muestra cómo reutilizar el código de preprocesamiento y los parámetros aprendidos durante el preprocesamiento antes de usar el modelo para la inferencia. Al reutilizar el código de preprocesamiento y los parámetros aprendidos, podremos asegurarnos de que las mismas transformaciones (como la normalización y la codificación de características) que se aplicaron a los datos de entrada durante el entrenamiento también se apliquen durante la inferencia. El modelo usado para la inferencia realizará predicciones sobre datos tabulares del conjunto de datos de enfermedades cardíacas de UCI.

Una visualización de la canalización es la siguiente:

El ejemplo de este artículo se basa en ejemplos de código incluidos en el repositorio azureml-examples. Para ejecutar los comandos localmente sin tener que copiar o pegar YAML y otros archivos, use los siguientes comandos para clonar el repositorio e ir a la carpeta del lenguaje de codificación:

CLI de Azure
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

Los archivos de este ejemplo están en:

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

Sigue estos pasos en los cuadernos de Jupyter Notebooks

Puedes seguir la versión SDK de Python de este ejemplo abriendo el cuaderno sdk-deploy-and-test.ipynb en el repositorio clonado.

Requisitos previos

Suscripción a Azure. Si no tiene una suscripción a Azure, cree una cuenta gratuita antes de empezar.
Un área de trabajo de Azure Machine Learning. Para crear un área de trabajo, vea Administración de áreas de trabajo de Azure Machine Learning.
Los siguientes permisos en el área de trabajo de Azure Machine Learning:
- Para crear o administrar puntos de conexión e implementaciones: utilice un rol de Propietario, Colaborador o personalizado al que se le hayan asignado los permisos Microsoft.MachineLearningServices/workspaces/batchEndpoints/*.
- Para crear implementaciones de Azure Resource Manager en el grupo de recursos del área de trabajo: use un rol de Propietario, Colaborador o personalizado que tenga asignado el permiso Microsoft.Resources/deployments/write en el grupo de recursos donde se implementa el área de trabajo.
La CLI de Azure Machine Learning o el SDK de Azure Machine Learning para Python:
- CLI de Azure
- Python
Ejecute el siguiente comando para instalar la CLI de Azure y la extensión ml para Azure Machine Learning:
```
az extension add -n ml
```
Las implementaciones de componentes de canalización para puntos de conexión por lotes se introdujeron en la versión 2.7 de la extensión ml para la CLI de Azure. Use el comando az extension update --name ml para obtener la versión más reciente.
Ejecute el siguiente comando para instalar el SDK de Azure Machine Learning para Python:
```
pip install azure-ai-ml
```
Las clases ModelBatchDeployment y PipelineComponentBatchDeployment se presentaron en la versión 1.7.0 del SDK. Use el comando pip install -U azure-ai-ml para obtener la versión más reciente.

Conexión con su área de trabajo

El área de trabajo es el recurso de nivel superior de Azure Machine Learning. Proporciona un lugar centralizado para trabajar con todos los artefactos que cree al usar Azure Machine Learning. En esta sección, se conectará al área de trabajo donde realizará las tareas de implementación.

CLI de Azure
Python

En el siguiente comando, escriba el id. de suscripción, el nombre del área de trabajo, el nombre de grupo de recursos y la ubicación:

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

Importe las bibliotecas necesarias:

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

Configura los detalles del área de trabajo y obtén un manipulador para el área de trabajo:

En el siguiente comando, escriba el id. de suscripción, el nombre del grupo de recursos y el nombre del área de trabajo:
```
subscription_id = "<subscription>"
resource_group = "<resource-group>"
workspace = "<workspace>"

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

Crear la canalización de inferencia

En esta sección, crearemos todos los recursos necesarios para nuestra canalización de inferencia. Comenzaremos creando un entorno que incluya las bibliotecas necesarias para los componentes de la canalización. A continuación, crearemos un clúster de proceso en el que se ejecutará la implementación por lotes. Después, registraremos los componentes, los modelos y las transformaciones necesarios para crear nuestra canalización de inferencia. Por último, compilaremos y probaremos la canalización.

Creación del entorno

Los componentes de este ejemplo utilizarán un entorno con las bibliotecas XGBoost y scikit-learn. El archivo environment/conda.yml contiene la configuración de la implementación:

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

Crear el entorno de la siguiente manera:

Definir el entorno:

CLI de Azure
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",
)

Crear el entorno:

CLI de Azure
Python

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

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

Creación de un clúster de proceso

Las implementaciones y puntos de conexión por lotes se ejecutan en clústeres de computación. Pueden ejecutarse en cualquier clúster de proceso de Azure Machine Learning que ya exista en el área de trabajo. Por lo tanto, varias implementaciones por lotes pueden compartir la misma infraestructura de proceso. En este ejemplo, trabajaremos en un clúster de proceso de Azure Machine Learning denominado batch-cluster. Verifiquemos que el cálculo exista en el área de trabajo o creémoslo de otra manera.

CLI de Azure
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()

Registro de componentes y modelos

Registraremos los componentes, modelos y transformaciones que necesitamos para crear nuestra canalización de inferencia. Podemos reutilizar algunos de estos recursos para las rutinas de entrenamiento.

Sugerencia

En este tutorial, volveremos a usar el modelo y el componente de preprocesamiento de una canalización de entrenamiento anterior. Para ver cómo se crearon, siga el ejemplo Cómo implementar una canalización de entrenamiento con puntos de conexión por lotes.

Registre el modelo que se vaya a usar para la predicción:

CLI de Azure
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)
)

El modelo registrado no se entrenó directamente en los datos de entrada. En su lugar, el dato de entrada se preprocesó (o transformó) antes del entrenamiento mediante un componente de preparación. También tendremos que registrar este componente. Registre el componente de preparación:
- CLI de Azure
- 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)
```
Sugerencia

Después de registrar el componente de preparación, ahora podrá hacer referencia a él desde el área de trabajo. Por ejemplo, azureml:uci_heart_prepare@latest obtendrá la última versión del componente de preparación.
Como parte de las transformaciones de datos del componente de preparación, los datos de entrada se normalizaron para centrar los predictores y limitar sus valores en el intervalo de [-1, 1]. Los parámetros de transformación se capturaron en una transformación Scikit-learn, que también se puede registrar para aplicarse más adelante cuando tengamos nuevos datos. Registre la transformación de la siguiente manera:
- CLI de Azure
- 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,
    )
)
```
Realizaremos inferencias para el modelo registrado mediante otro componente denominado score que calcula las predicciones de un modelo determinado. Se hará referencia al componente directamente desde su definición.

Sugerencia

El procedimiento recomendado sería registrar el componente y hacer referencia a él desde la canalización. Sin embargo, en este ejemplo vamos a hacer referencia al componente directamente desde su definición para ayudarle a ver qué componentes se reutilizan de la canalización de entrenamiento y cuáles son nuevos.

Crear la canalización

Ahora es el momento de enlazar todos los elementos juntos. La canalización de inferencia que implementaremos tiene dos componentes (pasos):

preprocess_job: este paso lee los datos de entrada y devuelve los datos preparados y las transformaciones aplicadas. El paso recibe dos entradas:
- data: una carpeta que contiene los datos de entrada que se van a puntuar
- transformations: (opcional) ruta de acceso a las transformaciones que se aplicarán, si están disponibles. Cuando se proporciona, las transformaciones se leen del modelo que se indica en la ruta de acceso. Sin embargo, si no se proporcionase la ruta de acceso, las transformaciones se aprenderán de los datos de entrada. Sin embargo, para la inferencia no es posible aprender los parámetros de transformación (en este ejemplo, los coeficientes de normalización) de los datos de entrada porque necesita usar los mismos valores de parámetro que se aprendieron durante el entrenamiento. Dado que esta entrada es opcional, el componente preprocess_job se puede usar durante el entrenamiento y la puntuación.
score_job: este paso realizará la inferencia en los datos transformados mediante el modelo de entrada. Observe que el componente usa un modelo de MLflow para realizar la inferencia. Por último, las puntuaciones se escriben en el mismo formato que se leyeron.

CLI de Azure
Python

La configuración de la canalización se define en el archivo 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")

Vamos a compilar la canalización en una función:

@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 visualización de la canalización es la siguiente:

Prueba de la canalización

Probemos la canalización con algunos datos de ejemplo. Para ello, se creará un trabajo mediante la canalización y el clúster de proceso batch-cluster creados anteriormente.

CLI de Azure
Python

El siguiente archivo pipeline-job.yml contiene la configuración del trabajo de canalización:

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

Ahora configuraremos algunos parámetros de ejecución para ejecutar la prueba:

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

Crear el trabajo de prueba:

CLI de Azure
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

Creación de un punto de conexión por lotes

Proporcione un nombre para el punto de conexión. El nombre de un punto de conexión por lotes debe ser único en cada región, ya que el nombre se usa para construir el URI de invocación. Para garantizar la unicidad, anexa los caracteres finales al nombre especificado en el código siguiente.
- CLI de Azure
- Python
```
ENDPOINT_NAME="uci-classifier-score"
```
```
endpoint_name = "uci-classifier-score"
```

Configuración del punto de conexión:

CLI de Azure
Python

El archivo endpoint.yml contiene la configuración del punto de conexión.

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

Creación del punto de conexión:

CLI de Azure
Python

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

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

Consulta del URI del punto de conexión:

CLI de Azure
Python

az ml batch-endpoint show --name $ENDPOINT_NAME

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

Implementar el componente de canalización

Para implementar el componente de canalización, tendremos que crear una implementación por lotes. Una implementación es un conjunto de recursos necesarios para hospedar el recurso que realiza el trabajo real.

Configurar la implementación

CLI de Azure
Python

El archivo deployment.yml contiene la configuración de la implementación. Puede comprobar el esquema YAML del punto de conexión por lotes completo para obtener más propiedades.

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

Nuestra canalización se define en una función. Para transformarlo en un componente, usará su propiedad component. Los componentes de canalización son gráficos de proceso reutilizables que se pueden incluir en implementaciones por lotes o que se pueden usar para crear canalizaciones más complejas.

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

Ahora podemos definir la implementación:

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

Creación de la implementación
- CLI de Azure
- Python
Ejecute el código siguiente para crear una implementación por lotes en el punto de conexión por lotes y establézcala como la implementación predeterminada.
```
az ml batch-deployment create --endpoint $ENDPOINT_NAME -f deployment.yml --set-default
```
Sugerencia

Observa el uso de la marca --set-default para indicar que esta nueva implementación es ahora el valor predeterminado.
Este comando iniciará la creación de la implementación y devolverá una respuesta de confirmación mientras continúa la creación de la implementación.
```
ml_client.batch_deployments.begin_create_or_update(deployment).result()
```
Una vez creada, vamos a configurar esta nueva implementación como la predeterminada:
```
endpoint = ml_client.batch_endpoints.get(endpoint_name)
endpoint.defaults.deployment_name = deployment.name
ml_client.batch_endpoints.begin_create_or_update(endpoint).result()
```
La implementación está lista para su uso.

Prueba de la implementación

Una vez creada la implementación, está lista para recibir trabajos. Siga estos pasos para probar lo siguiente:

Nuestra implementación requiere que indiquemos una entrada de datos y una entrada literal.
- CLI de Azure
- Python
El archivo inputs.yml contiene la definición del recurso de datos de entrada:

inputs.yml
```
inputs:
  input_data:
    type: uri_folder
    path: data/unlabeled
  score_mode:
    type: string
    default: append
outputs:
  scores:
    type: uri_folder
    mode: upload
```
La definición del recurso de datos de entrada:
```
input_data = Input(type="uri_folder", path="data/unlabeled/")
score_mode = Input(type="string", default="append")
```
Sugerencia

Para más información sobre cómo indicar entradas, consulta Crear trabajos y datos de entrada para puntos de conexión por lotes.
Puedes invocar la implementación determinada de la siguiente manera:
- CLI de Azure
- Python
```
JOB_NAME=$(az ml batch-endpoint invoke -n $ENDPOINT_NAME --f inputs.yml --query name -o tsv)
```
Sugerencia

¿Cuál es la diferencia entre el parámetro inputs y input al invocar un punto de conexión?

En general, puede usar un parámetro de diccionario inputs = {} con el método invoke para proporcionar un número arbitrario de entradas necesarias a un punto de conexión por lotes que contenga una implementación de modelo o una implementación de canalización.

Para una implementación de modelo, puede usar el parámetro input como una forma más corta de especificar la ubicación de los datos de entrada para la implementación. Este enfoque funciona porque una implementación de modelo siempre toma una entrada de datos.
```
job = ml_client.batch_endpoints.invoke(
    endpoint_name=endpoint.name,
    inputs={"input_data": input_data, "score_mode": score_mode},
)
```
Puedes supervisar el progreso de la presentación y transmitir los registros mediante:
- CLI de Azure
- Python
```
az ml job stream -n $JOB_NAME
```
```
ml_client.jobs.get(job.name)
```
Para esperar a que finalice el trabajo, ejecute el código siguiente:
```
ml_client.jobs.stream(name=job.name)
```

Acceso a la salida del trabajo

Una vez que se complete el trabajo, podemos acceder a su salida. Este trabajo contiene solo una salida denominada scores:

CLI de Azure
Python

Puede descargar los resultados asociados mediante az ml job download.

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

Descargue el resultado:

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

Lea los datos puntuados:

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

La salida tiene el siguiente aspecto:

age	sex	…	thal	prediction
0,9338	1	…	2	0
1,3782	1	…	3	1
1,3782	1	…	4	0
-1,954	1	…	3	0

La salida contiene las predicciones más los datos que se proporcionaron al componente de puntuación que se preprocesó. Por ejemplo, la columna age se normalizó y la columna thal contiene valores de codificación originales. En la práctica, probablemente desee generar la predicción solo y, a continuación, concatenarla con los valores originales. Este trabajo se ha dejado al lector.

Limpieza de recursos

Una vez que hayas terminado, elimina los recursos asociados del área de trabajo:

CLI de Azure
Python

Ejecuta el código siguiente para eliminar el punto de conexión por lotes y la implementación subyacente. --yes se usa para confirmar la eliminación.

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

Eliminación del punto de conexión:

ml_client.batch_endpoints.begin_delete(endpoint_name)

(Opcional) elimina el proceso, a menos que planees reutilizar el clúster de proceso con implementaciones posteriores.

CLI de Azure
Python

az ml compute delete -n batch-cluster

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

Compartir vía