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

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 de forma local sin tener que copiar/pegar YAML y otros archivos, primero clona el repositorio y luego cambia los directorios a la carpeta:

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

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.

Prerrequisitos

Antes de seguir los pasos de este artículo, asegúrese de que tiene los siguientes requisitos previos:

• Suscripción a Azure. Si no tiene una suscripción de Azure, cree una cuenta gratuita antes de empezar. Pruebe la versión gratuita o de pago de Azure Machine Learning.

• Un área de trabajo de Azure Machine Learning. Si no tiene una, siga los pasos descritos en el artículo Administración de áreas de trabajo de Azure Machine Learning para crear una.

• Asegúrese de tener los permisos siguientes en el área de trabajo:

  • Crear o administrar puntos de conexión e implementaciones por lotes: use un rol propietario, colaborador o personalizado que permita Microsoft.MachineLearningServices/workspaces/batchEndpoints/*.

  • Creación de implementaciones de ARM en el grupo de recursos del área de trabajo: use un rol propietario, colaborador o personalizado que permita Microsoft.Resources/deployments/write en el grupo de recursos donde se implementa el área de trabajo.

• Debe instalar el siguiente software para trabajar con Azure Machine Learning:

  CLI de Azure

  La CLI de Azure y la mlextensión para Azure Machine Learning.

  az extension add -n ml
  

  Nota

  Las implementaciones de componentes de canalización para puntos de conexión de Batch se introdujeron en la versión 2.7 de la extensión ml para la CLI de Azure. Use az extension update --name ml para obtener la última versión.

  Python

  El SDK de Azure Machine Learning para Python.

  pip install azure-ai-ml
  

  Nota

  Las clases ModelBatchDeployment y PipelineComponentBatchDeployment se introdujeron en la versión 1.7.0 del SDK. Use pip install -U azure-ai-ml para obtener la última versión.

Conexión con su área de trabajo

El área de trabajo es el recurso de nivel superior para Azure Machine Learning, que proporciona un lugar centralizado para trabajar con todos los artefactos que crea al usar Azure Machine Learning. En esta sección, nos conectaremos al área de trabajo en la que realizará las tareas de implementación.

CLI de Azure

Pasa los valores del id. de suscripción, el área de trabajo, la ubicación y el grupo de recursos en el código siguiente:

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

Python

1. 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
   

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

   Pase los valores del id. de suscripción, el área de trabajo y el grupo de recursos en el código siguiente:

   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:

1. Definir el entorno:

   CLI de 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. Crear el entorno:

   CLI de Azure

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

   Python

   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

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

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.

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

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

   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)

   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.

3. 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

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

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

Python

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

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

Python

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

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

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

1. 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

   ENDPOINT_NAME="uci-classifier-score"
   

   Python

   endpoint_name = "uci-classifier-score"

2. Configuración del punto de conexión:

   CLI de Azure

   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
   

   Python

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

3. Creación del punto de conexión:

   CLI de Azure

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

   Python

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

4. Consulta del URI del punto de conexión:

   CLI de Azure

   az ml batch-endpoint show --name $ENDPOINT_NAME
   

   Python

   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.

1. Configurar la implementación

   CLI de Azure

   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
   

   Python

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

2. Creación de la implementación

   CLI de Azure

   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.

   Python

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

3. 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:

1. Nuestra implementación requiere que indiquemos una entrada de datos y una entrada literal.

   CLI de Azure

   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
   

   Python

   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.

2. Puedes invocar la implementación determinada de la siguiente manera:

   CLI de Azure

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

   Python

   Sugerencia

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

   En general, se puede usar un diccionario inputs = {} con el método invoke para proporcionar un número arbitrario de entradas necesarias para 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 input como una manera más corta de especificar la ubicación de datos de entrada para la implementación, ya que una implementación del modelo siempre toma una sola entrada de datos.

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

3. Puedes supervisar el progreso de la presentación y transmitir los registros mediante:

   CLI de Azure

   az ml job stream -n $JOB_NAME
   

   Python

   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

Puede descargar los resultados asociados mediante az ml job download.

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

Python

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

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

Python

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

az ml compute delete -n batch-cluster

Python

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

Pasos siguientes

• Crear puntos de conexión por lotes a partir de trabajos de canalización
• Acceso a datos desde trabajos de puntos de conexión por lotes
• Solución de problemas de puntos de conexión por lotes