Como implantar um pipeline para executar a pontuação em lote com pré-processamento

Artigo
09/01/2024

APLICA-SE A:Azure CLI ml extension v2 (current)Python SDK azure-ai-ml v2 (current)

Neste artigo, você aprenderá como implantar um pipeline de inferência (ou pontuação) sob um ponto de extremidade em lote. O pipeline executa a pontuação em um modelo registrado enquanto também reutiliza um componente de pré-processamento de quando o modelo foi treinado. A reutilização do mesmo componente de pré-processamento garante que o mesmo pré-processamento seja aplicado durante a pontuação.

Ficará a saber como:

Criar um pipeline que reutilize componentes existentes do espaço de trabalho
Implantar o pipeline em um ponto de extremidade
Consumir previsões geradas pelo pipeline

Sobre este exemplo

Este exemplo mostra como reutilizar o código de pré-processamento e os parâmetros aprendidos durante o pré-processamento antes de usar seu modelo para inferência. Ao reutilizar o código de pré-processamento e os parâmetros aprendidos, podemos garantir que as mesmas transformações (como normalização e codificação de recursos) que foram aplicadas aos dados de entrada durante o treinamento também sejam aplicadas durante a inferência. O modelo utilizado para inferência realizará predições em dados tabulares do UCI Heart Disease Data set.

Uma visualização do pipeline é a seguinte:

O exemplo neste artigo é baseado em exemplos de código contidos no repositório azureml-examples . Para executar os comandos localmente sem ter que copiar ou colar YAML e outros arquivos, use os seguintes comandos para clonar o repositório e vá para a pasta para sua linguagem de codificação:

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

Os arquivos para este exemplo estão em:

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

Acompanhe nos cadernos Jupyter

Você pode acompanhar a versão do SDK do Python deste exemplo abrindo o bloco de anotações sdk-deploy-and-test.ipynb no repositório clonado.

Pré-requisitos

Uma subscrição do Azure. Se não tiver uma subscrição do Azure, crie uma conta gratuita antes de começar.
Uma área de trabalho do Azure Machine Learning. Para criar um espaço de trabalho, consulte Gerenciar espaços de trabalho do Azure Machine Learning.
As seguintes permissões no espaço de trabalho do Azure Machine Learning:
- Para criar ou gerenciar pontos de extremidade e implantações em lote: use um Proprietário, Colaborador ou função personalizada que tenha recebido as Microsoft.MachineLearningServices/workspaces/batchEndpoints/* permissões.
- Para criar implantações do Azure Resource Manager no grupo de recursos do espaço de trabalho: use um Proprietário, Colaborador ou função personalizada à qual tenha sido atribuída a Microsoft.Resources/deployments/write permissão no grupo de recursos em que o espaço de trabalho está implantado.
A CLI do Azure Machine Learning ou o SDK do Azure Machine Learning para Python:
- CLI do Azure
- Python
Execute o seguinte comando para instalar a CLI do Azure e a extensão para o ml Azure Machine Learning:
```
az extension add -n ml
```
As implantações de componentes de pipeline para pontos de extremidade em lote são introduzidas na versão 2.7 da ml extensão para a CLI do Azure. Use o az extension update --name ml comando para obter a versão mais recente.
Execute o seguinte comando para instalar o SDK do Azure Machine Learning para Python:
```
pip install azure-ai-ml
```
As ModelBatchDeployment classes e PipelineComponentBatchDeployment são introduzidas na versão 1.7.0 do SDK. Use o pip install -U azure-ai-ml comando para obter a versão mais recente.

Ligar à sua área de trabalho

O espaço de trabalho é o recurso de nível superior para o Azure Machine Learning. Ele fornece um local centralizado para trabalhar com todos os artefatos que você cria quando usa o Azure Machine Learning. Nesta seção, você se conecta ao espaço de trabalho onde executa suas tarefas de implantação.

CLI do Azure
Python

No comando a seguir, insira sua ID de assinatura, nome do espaço de trabalho, nome do grupo de recursos e local:

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

Importe as bibliotecas necessárias:

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

Configure os detalhes do espaço de trabalho e obtenha um identificador para o espaço de trabalho:

No comando a seguir, insira a ID da assinatura, o nome do grupo de recursos e o nome do espaço de trabalho:
```
subscription_id = "<subscription>"
resource_group = "<resource-group>"
workspace = "<workspace>"

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

Criar o pipeline de inferência

Nesta seção, criaremos todos os ativos necessários para nosso pipeline de inferência. Começaremos criando um ambiente que inclua as bibliotecas necessárias para os componentes do pipeline. Em seguida, criaremos um cluster de computação no qual a implantação em lote será executada. Depois, registraremos os componentes, modelos e transformações de que precisamos para construir nosso pipeline de inferência. Finalmente, vamos construir e testar o pipeline.

Criar o ambiente

Os componentes neste exemplo usarão um ambiente com as XGBoost bibliotecas e scikit-learn . O environment/conda.yml arquivo contém a configuração do ambiente:

ambiente/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

Crie o ambiente da seguinte maneira:

Defina o ambiente:

CLI do Azure
Python

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

Crie o ambiente:

CLI do Azure
Python

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

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

Criar um cluster de computação

Pontos de extremidade em lote e implantações são executados em clusters de computação. Eles podem ser executados em qualquer cluster de computação do Azure Machine Learning que já exista no espaço de trabalho. Portanto, várias implantações em lote podem compartilhar a mesma infraestrutura de computação. Neste exemplo, trabalharemos em um cluster de computação do Azure Machine Learning chamado batch-cluster. Vamos verificar se a computação existe no espaço de trabalho ou criá-la de outra forma.

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

Registar componentes e modelos

Vamos registrar componentes, modelos e transformações de que precisamos para construir nosso pipeline de inferência. Podemos reutilizar alguns destes ativos para rotinas de treino.

Gorjeta

Neste tutorial, reutilizaremos o modelo e o componente de pré-processamento de um pipeline de treinamento anterior. Você pode ver como eles foram criados seguindo o exemplo Como implantar um pipeline de treinamento com pontos de extremidade em lote.

Registre o modelo a ser usado para previsão:

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

O modelo registrado não foi treinado diretamente em dados de entrada. Em vez disso, os dados de entrada foram pré-processados (ou transformados) antes do treinamento, usando um componente de preparação. Também precisaremos registrar esse componente. Registre o componente de preparação:
- CLI do 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)
```
Gorjeta

Depois de registrar o componente de preparação, agora você pode consultá-lo a partir do espaço de trabalho. Por exemplo, azureml:uci_heart_prepare@latest obterá a última versão do componente de preparação.
Como parte das transformações de dados no componente de preparação, os dados de entrada foram normalizados para centralizar os preditores e limitar seus valores no intervalo de [-1, 1]. Os parâmetros de transformação foram capturados em uma transformação scikit-learn que também podemos registrar para aplicar mais tarde quando tivermos novos dados. Registe a transformação da seguinte forma:
- CLI do 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 inferências para o modelo registrado, usando outro componente chamado score que calcula as previsões para um determinado modelo. Referenciaremos o componente diretamente de sua definição.

Gorjeta

A melhor prática seria registrar o componente e fazer referência a ele a partir do pipeline. No entanto, neste exemplo, vamos fazer referência ao componente diretamente de sua definição para ajudá-lo a ver quais componentes são reutilizados do pipeline de treinamento e quais são novos.

Construa o pipeline

Agora é hora de unir todos os elementos. O pipeline de inferência que implantaremos tem dois componentes (etapas):

preprocess_job: Esta etapa lê os dados de entrada e retorna os dados preparados e as transformações aplicadas. A etapa recebe duas entradas:
- data: uma pasta que contém os dados de entrada para pontuar
- transformations: (opcional) Caminho para as transformações que serão aplicadas, se disponíveis. Quando fornecidas, as transformações são lidas a partir do modelo indicado no caminho. No entanto, se o caminho não for fornecido, as transformações serão aprendidas a partir dos dados de entrada. Para inferir, no entanto, você não pode aprender os parâmetros de transformação (neste exemplo, os coeficientes de normalização) a partir dos dados de entrada porque você precisa usar os mesmos valores de parâmetros que foram aprendidos durante o treinamento. Como essa entrada é opcional, o componente pode ser usado durante o preprocess_job treinamento e a pontuação.
score_job: Esta etapa realizará a inferência nos dados transformados, usando o modelo de entrada. Observe que o componente usa um modelo MLflow para executar inferência. Finalmente, as partituras são escritas no mesmo formato em que foram lidas.

CLI do Azure
Python

A configuração do pipeline é definida no pipeline.yml arquivo:

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 construir o pipeline em uma função:

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

Uma visualização do pipeline é a seguinte:

Testar o pipeline

Vamos testar o pipeline com alguns dados de exemplo. Para fazer isso, criaremos um trabalho usando o pipeline e o cluster de batch-cluster computação criado anteriormente.

CLI do Azure
Python

O arquivo a seguir pipeline-job.yml contém a configuração para o trabalho de 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

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

Agora, vamos definir algumas configurações de execução para executar o teste:

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

Crie o trabalho de teste:

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

Criar um ponto de extremidade em lote

Forneça um nome para o ponto de extremidade. O nome de um ponto de extremidade em lote precisa ser exclusivo em cada região, pois o nome é usado para construir o URI de invocação. Para garantir a exclusividade, acrescente quaisquer caracteres à direita ao nome especificado no código a seguir.
- CLI do Azure
- Python
```
ENDPOINT_NAME="uci-classifier-score"
```
```
endpoint_name = "uci-classifier-score"
```

Configure o ponto de extremidade:

CLI do Azure
Python

O endpoint.yml arquivo contém a configuração do ponto de extremidade.

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

Crie o ponto de extremidade:

CLI do Azure
Python

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

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

Consulte o URI do ponto de extremidade:

CLI do Azure
Python

az ml batch-endpoint show --name $ENDPOINT_NAME

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

Implantar o componente de pipeline

Para implantar o componente de pipeline, temos que criar uma implantação em lote. Uma implantação é um conjunto de recursos necessários para hospedar o ativo que faz o trabalho real.

Configurar a implementação

CLI do Azure
Python

O deployment.yml arquivo contém a configuração da implantação. Você pode verificar o esquema YAML do ponto de extremidade de lote completo para obter propriedades extras.

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

Nosso pipeline é definido em uma função. Para transformá-lo em um componente, você usará a component propriedade dele. Os componentes de pipeline são gráficos de computação reutilizáveis que podem ser incluídos em implantações em lote ou usados para compor pipelines mais complexos.

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

Agora podemos definir a implantação:

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

Criar a implantação
- CLI do Azure
- Python
Execute o código a seguir para criar uma implantação em lote no ponto de extremidade em lote e defini-la como a implantação padrão.
```
az ml batch-deployment create --endpoint $ENDPOINT_NAME -f deployment.yml --set-default
```
Gorjeta

Observe o --set-default uso do sinalizador para indicar que essa nova implantação agora é o padrão.
Esse comando iniciará a criação da implantação e retornará uma resposta de confirmação enquanto a criação da implantação continuar.
```
ml_client.batch_deployments.begin_create_or_update(deployment).result()
```
Uma vez criada, vamos configurar essa nova implantação como padrão:
```
endpoint = ml_client.batch_endpoints.get(endpoint_name)
endpoint.defaults.deployment_name = deployment.name
ml_client.batch_endpoints.begin_create_or_update(endpoint).result()
```
Sua implantação está pronta para uso.

Testar a implementação

Depois que a implantação é criada, ela está pronta para receber trabalhos. Siga estas etapas para testá-lo:

Nossa implantação requer que indiquemos uma entrada de dados e uma entrada literal.
- CLI do Azure
- Python
O inputs.yml arquivo contém a definição para o ativo de dados 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
```
A definição de ativo de dados de cálculo:
```
input_data = Input(type="uri_folder", path="data/unlabeled/")
score_mode = Input(type="string", default="append")
```
Gorjeta

Para saber mais sobre como indicar entradas, consulte Criar trabalhos e dados de entrada para pontos de extremidade em lote.
Você pode invocar a implantação padrão da seguinte maneira:
- CLI do Azure
- Python
```
JOB_NAME=$(az ml batch-endpoint invoke -n $ENDPOINT_NAME --f inputs.yml --query name -o tsv)
```
Gorjeta

Qual é a diferença entre o inputs parâmetro e input quando você invoca um ponto de extremidade?

Em geral, você pode usar um parâmetro de dicionário inputs = {} com o invoke método para fornecer um número arbitrário de entradas necessárias para um ponto de extremidade em lote que contém uma implantação de modelo ou uma implantação de pipeline.

Para uma implantação de modelo, você pode usar o input parâmetro como uma maneira mais curta de especificar o local dos dados de entrada para a implantação. Essa abordagem funciona porque uma implantação de modelo sempre usa apenas uma entrada de dados.
```
job = ml_client.batch_endpoints.invoke(
    endpoint_name=endpoint.name,
    inputs={"input_data": input_data, "score_mode": score_mode},
)
```
Você pode monitorar o progresso do show e transmitir os logs usando:
- CLI do Azure
- Python
```
az ml job stream -n $JOB_NAME
```
```
ml_client.jobs.get(job.name)
```
Para aguardar a conclusão do trabalho, execute o seguinte código:
```
ml_client.jobs.stream(name=job.name)
```

Acessar saída de trabalho

Uma vez que o trabalho é concluído, podemos acessar sua saída. Este trabalho contém apenas uma saída chamada scores:

CLI do Azure
Python

Você pode baixar os resultados associados usando az ml job downloado .

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

Descarregue o resultado:

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

Leia os dados da pontuação:

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

A saída tem a seguinte aparência:

idade	sexo	...	Thal	previsão
0.9338	1	...	2	0
1.3782	1	...	3	1
1.3782	1	...	4	0
-1.954	1	...	3	0

A saída contém as previsões mais os dados que foram fornecidos ao componente de pontuação , que foi pré-processado. Por exemplo, a coluna age foi normalizada e a coluna thal contém valores de codificação originais. Na prática, você provavelmente deseja produzir apenas a previsão e, em seguida, combiná-la com os valores originais. Esta obra foi deixada ao leitor.

Clean up resources (Limpar recursos)

Quando terminar, exclua os recursos associados do espaço de trabalho:

CLI do Azure
Python

Execute o código a seguir para excluir o ponto de extremidade em lote e sua implantação subjacente. --yes é utilizado para confirmar a eliminação.

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

Exclua o ponto de extremidade:

ml_client.batch_endpoints.begin_delete(endpoint_name)

(Opcional) Exclua a computação, a menos que planeje reutilizar seu cluster de computação com implantações posteriores.

CLI do Azure
Python

az ml compute delete -n batch-cluster

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

Partilhar via