Creación y ejecución de canalizaciones de aprendizaje automático mediante componentes con el SDK de Azure Machine Learning v2

SE APLICA A: SDK de Python azure-ai-ml v2 (actual)

En este artículo, aprenderá a crear una canalización de Azure Machine Learning mediante el SDK de Python v2 para completar una tarea de clasificación de imágenes que contiene tres pasos: preparar datos, entrenar un modelo de clasificación de imágenes y puntuar el modelo. Las canalizaciones de Machine Learning optimizan el flujo de trabajo con velocidad, portabilidad y reutilización, con el fin de que pueda centrarse en el aprendizaje automático, en lugar de en la infraestructura y la automatización.

En el ejemplo se entrena una pequeña red neuronal convolucional de Keras para clasificar las imágenes del conjunto de datos de MNIST Fashion. La canalización es similar a la siguiente.

En este artículo, se realizarán las siguientes tareas:

• Preparar los datos de entrada para el trabajo de canalización
• Crear tres componentes para preparar los datos, entrenar y puntuar
• Crear una canalización desde los componentes
• Obtener acceso al área de trabajo con proceso
• Enviar el trabajo de canalización
• Revisar la salida de los componentes y de la red neuronal entrenada
• (Opcional) Registrar el componente para reutilizarlo y compartirlo en el área de trabajo

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

Requisitos previos

• Área de trabajo de Azure Machine Learning: si no tiene una, complete el tutorial creación de recursos.

• Un entorno de Python en el que ha instalado el SDK de Azure Machine Learning v2(instrucciones de instalación). Consulte la sección de introducción. Este entorno es para definir y controlar los recursos de Azure Machine Learning y es independiente del entorno que se usa en tiempo de ejecución para el entrenamiento.

• Clonación del repositorio de ejemplos

  Para ejecutar los ejemplos de entrenamiento, primero clone el repositorio de ejemplos y cambie al directorio sdk:

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

Inicio de una sesión interactiva de Python

En este artículo se usa el SDK de Python para Azure Machine Learning para crear y controlar una canalización de Azure Machine Learning. En el artículo se da por supuesto que va a ejecutar los fragmentos de código de forma interactiva en un entorno REPL de Python o en un cuaderno de Jupyter Notebook.

Este artículo se basa en el cuaderno image_classification_keras_minist_convnet.ipynb que se encuentra en el directorio sdk/python/jobs/pipelines/2e_image_classification_keras_minist_convnet del repositorio de ejemplos de Azure Machine Learning.

Importación de bibliotecas necesarias

Importe todas las bibliotecas requeridas de Azure Machine Learning que necesitará para este artículo:

# import required libraries
from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential

from azure.ai.ml import MLClient
from azure.ai.ml.dsl import pipeline
from azure.ai.ml import load_component

Preparación de los datos de entrada para el trabajo de canalización

Debe preparar los datos de entrada para esta canalización de clasificación de imágenes.

Fashion-MNIST es un conjunto de datos de imágenes de moda dividido en 10 clases. Cada imagen es una imagen en escala de grises de 28 x 28 y hay 60 000 imágenes de entrenamiento y 10 000 de prueba. Como problema de clasificación de imágenes, Fashion-MNIST es más compleja que la base de datos de dígitos manuscritos de MNIST clásica. Se distribuye en el mismo formato binario comprimido que la base de datos de dígitos manuscritos original.

Importe todas las bibliotecas requeridas de Azure Machine Learning que necesitará.

Si define un objeto Input, se crea una referencia a la ubicación de los orígenes de datos. Los datos se mantienen en la ubicación existente, por lo que no se genera ningún costo de almacenamiento adicional.

Creación de componentes para la canalización de compilación

La tarea de clasificación de imágenes se puede dividir en tres pasos: preparar los datos, entrenar el modelo y puntuar el modelo.

Un componente de Azure Machine Learning es un fragmento de código independiente que realiza un paso en una canalización de aprendizaje automático. En este artículo, creará tres componentes para la tarea de clasificación de imágenes:

• Preparación de los datos para el entrenamiento y la prueba
• Entrenamiento de una red neuronal para la clasificación de imágenes mediante datos de entrenamiento
• Puntuación del modelo con los datos de prueba

Para cada componente, debe preparar lo siguiente:

1. Preparación del script de Python que contiene la lógica de ejecución

2. Definición de la interfaz del componente

3. Agregue otros metadatos del componente, incluido el entorno de ejecución, el comando para ejecutar el componente, etc.

En la sección siguiente se mostrarán los componentes de creación de dos maneras diferentes: los dos primeros componentes mediante la función de Python y el tercer componente mediante la definición de YAML.

Creación del componente de preparación de datos

El primer componente de esta canalización convertirá los archivos de datos comprimidos de fashion_ds en dos archivos CSV, uno para el entrenamiento y el otro para la puntuación. Usará la función de Python para definir este componente.

Si sigue el ejemplo del repositorio de ejemplos de Azure Machine Learning, los archivos de código fuente ya están disponibles en la carpeta prep/. Esta carpeta contiene dos archivos para construir el componente: prep_component.py, que define el componente y conda.yaml, que define el entorno de ejecución del componente.

Definición del componente mediante la función de Python

Mediante el uso de la función command_component() como decorador, puede definir fácilmente la interfaz, los metadatos y el código del componente para ejecutarse desde una función de Python. Cada función de Python decorada se transformará en una única especificación estática (YAML) que el servicio de canalización puede procesar.

# Converts MNIST-formatted files at the passed-in input path to training data output path and test data output path
import os
from pathlib import Path
from mldesigner import command_component, Input, Output


@command_component(
    name="prep_data",
    version="1",
    display_name="Prep Data",
    description="Convert data to CSV file, and split to training and test data",
    environment=dict(
        conda_file=Path(__file__).parent / "conda.yaml",
        image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04",
    ),
)
def prepare_data_component(
    input_data: Input(type="uri_folder"),
    training_data: Output(type="uri_folder"),
    test_data: Output(type="uri_folder"),
):
    convert(
        os.path.join(input_data, "train-images-idx3-ubyte"),
        os.path.join(input_data, "train-labels-idx1-ubyte"),
        os.path.join(training_data, "mnist_train.csv"),
        60000,
    )
    convert(
        os.path.join(input_data, "t10k-images-idx3-ubyte"),
        os.path.join(input_data, "t10k-labels-idx1-ubyte"),
        os.path.join(test_data, "mnist_test.csv"),
        10000,
    )


def convert(imgf, labelf, outf, n):
    f = open(imgf, "rb")
    l = open(labelf, "rb")
    o = open(outf, "w")

    f.read(16)
    l.read(8)
    images = []

    for i in range(n):
        image = [ord(l.read(1))]
        for j in range(28 * 28):
            image.append(ord(f.read(1)))
        images.append(image)

    for image in images:
        o.write(",".join(str(pix) for pix in image) + "\n")
    f.close()
    o.close()
    l.close()

El código anterior define un componente con el nombre para mostrar Prep Data mediante el decorador @command_component:

• name es el identificador único del componente.

• version es la versión actual del componente. Un componente puede tener varias versiones.

• display_name es un nombre para mostrar descriptivo del componente en la interfaz de usuario, que no es único.

• description normalmente describe qué tarea puede completar este componente.

• environment especifica el entorno de ejecución para este componente. El entorno de este componente especifica una imagen de Docker y hace referencia al archivo conda.yaml.

  El archivo conda.yaml contiene todos los paquetes usados para el componente como el siguiente:

  name: imagekeras_prep_conda_env
  channels:
    - defaults
  dependencies:
    - python=3.7.11
    - pip=20.0
    - pip:
      - mldesigner==0.1.0b4
  

• La función prepare_data_component define una entrada para input_data y dos salidas para training_data y test_data. input_data es la ruta de acceso de datos de entrada. training_data y test_data son rutas de acceso de datos de salida para datos de entrenamiento y datos de prueba.

• Este componente convierte los datos de input_data en un csv de datos de entrenamiento para training_data y un csv de datos de prueba para test_data.

A continuación se muestra el aspecto de un componente en la interfaz de usuario de Estudio.

• Un componente es un bloque de un gráfico de canalización.
• input_data, training_data y test_data son puertos del componente, que se conecta a otros componentes para el streaming de datos.

Ahora, ha preparado todos los archivos de código fuente para el componente Prep Data.

Creación del componente de entrenamiento del modelo

En esta sección, creará un componente para entrenar el modelo de clasificación de imágenes en la función de Python, como el componente Prep Data.

La diferencia es que, dado que la lógica de entrenamiento es más complicada, puede colocar el código de entrenamiento original en un archivo de Python independiente.

Los archivos de código fuente de este componente se encuentran en la carpeta train/ del repositorio de ejemplos de Azure Machine Learning. Esta carpeta contiene tres archivos para construir el componente:

• train.py: contiene la lógica real para entrenar el modelo.
• train_component.py: define la interfaz del componente e importa la función en train.py.
• conda.yaml: define el entorno de ejecución del componente.

Obtención de un script que contiene la lógica de ejecución

El archivo train.py contiene una función normal de Python, que realiza la lógica del modelo de entrenamiento para entrenar una red neuronal Keras para la clasificación de imágenes. Para ver el código, consulte el archivo train.py en GitHub.

Definición del componente mediante la función de Python

Después de definir correctamente la función de entrenamiento, puede usar @command_component en el SDK v2 de Azure Machine Learning para encapsular la función como un componente, que se puede usar en canalizaciones de Azure Machine Learning.

import os
from pathlib import Path
from mldesigner import command_component, Input, Output


@command_component(
    name="train_image_classification_keras",
    version="1",
    display_name="Train Image Classification Keras",
    description="train image classification with keras",
    environment=dict(
        conda_file=Path(__file__).parent / "conda.yaml",
        image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04",
    ),
)
def keras_train_component(
    input_data: Input(type="uri_folder"),
    output_model: Output(type="uri_folder"),
    epochs=10,
):
    # avoid dependency issue, execution logic is in train() func in train.py file
    from train import train

    train(input_data, output_model, epochs)

El código anterior define un componente con el nombre para mostrar Train Image Classification Keras mediante @command_component:

• La función keras_train_component define una entrada input_data de la que proceden los datos de entrenamiento, una entrada epochs que especifica épocas durante el entrenamiento y una salida output_model en la que genera el archivo de modelo. El valor predeterminado de epochs es 10. La lógica de ejecución de este componente procede de la función train() en train.py anterior.

El componente de entrenamiento del modelo tiene una configuración ligeramente más compleja que el componente de preparación de datos. conda.yaml es similar a lo siguiente:

name: imagekeras_train_conda_env
channels:
  - defaults
dependencies:
  - python=3.8
  - pip=20.2
  - pip:
    - mldesigner==0.1.0b12
    - azureml-mlflow==1.50.0
    - tensorflow==2.7.0
    - numpy==1.21.4
    - scikit-learn==1.0.1
    - pandas==1.3.4
    - matplotlib==3.2.2
    - protobuf==3.20.0

Ahora, ha preparado todos los archivos de código fuente para el componente Train Image Classification Keras.

Creación del componente de puntuación del modelo

En esta sección, aparte de los componentes anteriores, creará un componente para puntuar el modelo entrenado a través de la especificación y el script de YAML.

Si sigue el ejemplo del repositorio de ejemplos de Azure Machine Learning, los archivos de código fuente ya están disponibles en la carpeta score/. Esta carpeta contiene tres archivos para construir el componente:

• score.py contiene el código fuente del componente.
• score.yaml: define la interfaz y otros detalles del componente.
• conda.yaml: define el entorno de ejecución del componente.

Obtención de un script que contiene la lógica de ejecución

El archivo score.py contiene una función normal de Python, que realiza la lógica del modelo de entrenamiento.

from tensorflow import keras
from keras.models import Sequential
from keras.layers import Dense, Dropout, Flatten
from keras.layers import Conv2D, MaxPooling2D
from tensorflow.keras.layers import BatchNormalization
from tensorflow.keras.utils import to_categorical
from keras.callbacks import Callback
from keras.models import load_model

import argparse
from pathlib import Path
import numpy as np
import pandas as pd
import os
import matplotlib.pyplot as plt
import mlflow


def get_file(f):

    f = Path(f)
    if f.is_file():
        return f
    else:
        files = list(f.iterdir())
        if len(files) == 1:
            return files[0]
        else:
            raise Exception("********This path contains more than one file*******")


def parse_args():
    # setup argparse
    parser = argparse.ArgumentParser()

    # add arguments
    parser.add_argument(
        "--input_data", type=str, help="path containing data for scoring"
    )
    parser.add_argument(
        "--input_model", type=str, default="./", help="input path for model"
    )

    parser.add_argument(
        "--output_result", type=str, default="./", help="output path for model"
    )

    # parse args
    args = parser.parse_args()

    # return args
    return args


def score(input_data, input_model, output_result):

    test_file = get_file(input_data)
    data_test = pd.read_csv(test_file, header=None)

    img_rows, img_cols = 28, 28
    input_shape = (img_rows, img_cols, 1)

    # Read test data
    X_test = np.array(data_test.iloc[:, 1:])
    y_test = to_categorical(np.array(data_test.iloc[:, 0]))
    X_test = (
        X_test.reshape(X_test.shape[0], img_rows, img_cols, 1).astype("float32") / 255
    )

    # Load model
    files = [f for f in os.listdir(input_model) if f.endswith(".h5")]
    model = load_model(input_model + "/" + files[0])

    # Log metrics of the model
    eval = model.evaluate(X_test, y_test, verbose=0)

    mlflow.log_metric("Final test loss", eval[0])
    print("Test loss:", eval[0])

    mlflow.log_metric("Final test accuracy", eval[1])
    print("Test accuracy:", eval[1])

    # Score model using test data
    y_predict = model.predict(X_test)
    y_result = np.argmax(y_predict, axis=1)

    # Output result
    np.savetxt(output_result + "/predict_result.csv", y_result, delimiter=",")


def main(args):
    score(args.input_data, args.input_model, args.output_result)


# run script
if __name__ == "__main__":
    # parse args
    args = parse_args()

    # call main function
    main(args)

El código de score.py toma tres argumentos de línea de comandos: input_data, input_model y output_result. El programa puntúa el modelo de entrada mediante datos de entrada y, a continuación, genera el resultado de puntuación.

Definición del componente a través de YAML

En esta sección, aprenderá a crear una especificación de componente en el formato válido de especificación del componente de YAML. Este archivo especifica la siguiente información:

• Metadatos: nombre, nombre para mostrar, versión, tipo, etc.
• Interfaz: entradas y salidas
• Comando, código y entorno: el comando, el código y el entorno usados para ejecutar el componente

$schema: https://azuremlschemas.azureedge.net/latest/commandComponent.schema.json
type: command

name: score_image_classification_keras
display_name: Score Image Classification Keras
inputs:
  input_data: 
    type: uri_folder
  input_model:
    type: uri_folder
outputs:
  output_result:
    type: uri_folder
code: ./
command: python score.py --input_data ${{inputs.input_data}} --input_model ${{inputs.input_model}} --output_result ${{outputs.output_result}}
environment:
  conda_file: ./conda.yaml
  image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04

• name es el identificador único del componente. Su nombre para mostrar es Score Image Classification Keras.
• Este componente tiene dos entradas y una salida.
• La ruta de acceso al código fuente se define en la sección code y cuando el componente se ejecuta en la nube, todos los archivos de esa ruta de acceso se cargarán como la instantánea de este componente.
• La sección command especifica el comando que se va a ejecutar mientras se ejecuta este componente.
• La sección environment contiene una imagen de Docker y un archivo YAML de Conda. El archivo de código fuente se encuentra en el repositorio de ejemplos.

Ahora, tiene todos los archivos de código fuente para el componente de puntuación del modelo.

Carga de componentes para la canalización de compilación

Para el componente de preparación de datos y el componente de entrenamiento de modelos definidos por la función de Python, puede importar los componentes igual que las funciones normales de Python.

En el siguiente código, se importan las funciones prepare_data_component() y keras_train_component() del archivo prep_component.py de la carpeta prep y del archivo train_component de la carpeta train, respectivamente.

%load_ext autoreload
%autoreload 2

# load component function from component python file
from prep.prep_component import prepare_data_component
from train.train_component import keras_train_component

# print hint of components
help(prepare_data_component)
help(keras_train_component)

Para el componente de puntuación definido por YAML, puede usar la función load_component() para cargar.

# load component function from yaml
keras_score_component = load_component(source="./score/score.yaml")

Compilación de la canalización

Ahora que ha creado y cargado todos los componentes y los datos de entrada para compilar la canalización, puede juntarlos en una canalización:

Nota:

Para usar el proceso sin servidor, agregue from azure.ai.ml.entities import ResourceConfiguration al principio. A continuación, reemplace:

• default_compute=cpu_compute_target, con default_compute="serverless",
• train_node.compute = gpu_compute_target con train_node.resources = "ResourceConfiguration(instance_type="Standard_NC6s_v3",instance_count=2)

# define a pipeline containing 3 nodes: Prepare data node, train node, and score node
@pipeline(
    default_compute=cpu_compute_target,
)
def image_classification_keras_minist_convnet(pipeline_input_data):
    """E2E image classification pipeline with keras using python sdk."""
    prepare_data_node = prepare_data_component(input_data=pipeline_input_data)

    train_node = keras_train_component(
        input_data=prepare_data_node.outputs.training_data
    )
    train_node.compute = gpu_compute_target

    score_node = keras_score_component(
        input_data=prepare_data_node.outputs.test_data,
        input_model=train_node.outputs.output_model,
    )


# create a pipeline
pipeline_job = image_classification_keras_minist_convnet(pipeline_input_data=mnist_ds)

La canalización tiene un proceso cpu_compute_target predeterminado, lo que significa que si no especifica el proceso para un nodo específico, ese nodo se ejecutará en el proceso predeterminado.

La canalización tiene una entrada pipeline_input_data de nivel de canalización. Puede asignar valor a la entrada de canalización al enviar un trabajo de canalización.

La canalización contiene tres nodos, prepare_data_node, train_node y score_node.

• El valor input_data de prepare_data_node utiliza el valor de pipeline_input_data.

• El valor input_data de train_node es de la salida training_data del nodo prepare_data_node.

• El valor input_data de score_node es de la salida test_data de prepare_data_node, y el valor input_model es del valor output_model de train_node.

• Dado que train_node entrenará un modelo de CNN, puede especificar su proceso como gpu_compute_target, lo que puede mejorar el rendimiento del entrenamiento.

Envío del trabajo de canalización

Ahora que ha creado la canalización, puede enviarla al área de trabajo. Para enviar un trabajo, primero debe conectarse a un área de trabajo.

Obtención de acceso al área de trabajo

Configuración de credenciales

Usaremos DefaultAzureCredential para acceder al área de trabajo. DefaultAzureCredential debe ser capaz de controlar la mayoría de los escenarios de autenticación del SDK de Azure.

Referencia para obtener más credenciales disponibles si no funciona: configuración del ejemplo de credenciales, documento de referencia de azure-identity.

try:
    credential = DefaultAzureCredential()
    # Check if given credential can get token successfully.
    credential.get_token("https://management.azure.com/.default")
except Exception as ex:
    # Fall back to InteractiveBrowserCredential in case DefaultAzureCredential not work
    credential = InteractiveBrowserCredential()

Obtención de un manipulador para un área de trabajo con proceso

Cree un objeto MLClient para administrar los servicios de Azure Machine Learning. Si usa el proceso sin servidor, no es necesario crear estos procesos.

# Get a handle to workspace
ml_client = MLClient.from_config(credential=credential)

# Retrieve an already attached Azure Machine Learning Compute.
cpu_compute_target = "cpu-cluster"
print(ml_client.compute.get(cpu_compute_target))
gpu_compute_target = "gpu-cluster"
print(ml_client.compute.get(gpu_compute_target))

Importante

Este fragmento de código espera que el archivo .json de configuración del área de trabajo se guarde en el directorio actual o en su elemento primario. Para obtener más información sobre la creación de un área de trabajo, consulte Creación de recursos para el área de trabajo. Para más información sobre cómo guardar la configuración en un archivo, consulte Creación de un archivo de configuración de área de trabajo.

Envío del trabajo de canalización al área de trabajo

Ahora que tiene un manipulador para el área de trabajo, puede enviar el trabajo de canalización.

pipeline_job = ml_client.jobs.create_or_update(
    pipeline_job, experiment_name="pipeline_samples"
)
pipeline_job

El código anterior envía este trabajo de canalización de clasificación de imágenes a un experimento llamado pipeline_samples. Si no existe, creará automáticamente el experimento. El objeto pipeline_input_data usa fashion_ds.

La llamada a pipeline_job genera una salida similar a la siguiente:

La llamada para submit el Experiment se completa rápidamente y produce una salida parecida a:

Experimento	Nombre	Tipo	Estado	Página de detalles
pipeline_samples	sharp_pipe_4gvqx6h1fb	pipeline	Preparing (En preparación)	Vínculo a Estudio de Azure Machine Learning.

Para supervisar la ejecución de la canalización, abra el vínculo o puede bloquearlo hasta que se complete mediante la ejecución de:

# wait until the job completes
ml_client.jobs.stream(pipeline_job.name)

Importante

La primera ejecución de canalización tarda aproximadamente 15 minutos. Se deben descargar todas las dependencias, se crea una imagen de Docker y se aprovisiona y se crea el entorno de Python. La repetición de la ejecución de la canalización tarda mucho menos, ya que esos recursos se reutilizan en lugar de crearse. Sin embargo, el tiempo de ejecución total de la canalización depende de la carga de trabajo de los scripts y de los procesos que se ejecutan en cada paso de la canalización.

Comprobación de salidas y depuración de la canalización en la interfaz de usuario

Puede abrir Link to Azure Machine Learning studio, que es la página de detalles del trabajo de la canalización. Verá el gráfico de canalización como se muestra a continuación.

Para comprobar los registros y salidas de cada componente, haga clic con el botón derecho en el componente o seleccione el componente para abrir su panel de detalles. Para más información sobre cómo depurar la canalización en la interfaz de usuario, consulte Cómo usar la depuración de un error de canalización.

(Opcional) Registro de componentes en el área de trabajo

En la sección anterior, ha creado una canalización con tres componentes para que E2E complete una tarea de clasificación de imágenes. También puede registrar componentes en el área de trabajo para que se puedan compartir y volver a usar en el área de trabajo. A continuación se muestra un ejemplo para registrar el componente de preparación de datos.

try:
    # try get back the component
    prep = ml_client.components.get(name="prep_data", version="1")
except:
    # if not exists, register component using following code
    prep = ml_client.components.create_or_update(prepare_data_component)

# list all components registered in workspace
for c in ml_client.components.list():
    print(c)

Con ml_client.components.get(), puede obtener un componente registrado por nombre y versión. Con ml_client.components.create_or_update(), puede registrar un componente cargado previamente desde la función de Python o YAML.

Pasos siguientes

• Para más ejemplos de cómo crear canalizaciones mediante el SDK de Machine Learning, consulte el repositorio de ejemplos.
• Para obtener información sobre cómo usar la interfaz de usuario de Estudio para enviar y depurar la canalización, consulte cómo crear canalizaciones mediante el componente en la interfaz de usuario.
• Para obtener información sobre cómo usar la CLI de Azure Machine Learning para crear componentes y canalizaciones, consulte cómo crear canalizaciones mediante el componente con la CLI.
• Para saber cómo implementar canalizaciones en la producción mediante puntos de conexión por lotes, consulta Cómo implementar canalizaciones con puntos de conexión por lotes.