Развертывание конвейера для выполнения пакетной оценки с предварительной обработкой

Статья
09/02/2024

ОБЛАСТЬ ПРИМЕНЕНИЯ:Расширение машинного обучения Azure CLI версии 2 (current)Python SDK azure-ai-ml версии 2 (current)

В этой статье вы узнаете, как развернуть конвейер вывода (или оценки) в конечной точке пакетной службы. Конвейер выполняет оценку по зарегистрированной модели, а также повторно использует компонент предварительной обработки, начиная с того момента, когда модель была обучена. Повторное использование одного и того же компонента предварительной обработки гарантирует применение той же предварительной обработки во время оценки.

Содержание:

Создание конвейера, который повторно использует существующие компоненты из рабочей области
Развертывание конвейера в конечной точке
Использование прогнозов, созданных конвейером

Сведения об этом примере

В этом примере показано, как повторно использовать код предварительной обработки и параметры, полученные во время предварительной обработки перед использованием модели для вывода. Повторно используя предварительно обработанный код и обучаемые параметры, мы можем гарантировать, что те же преобразования (например, нормализация и кодировка признаков), которые были применены к входным данным во время обучения, также применяются во время вывода. Модель, используемая для вывода, будет выполнять прогнозирование табличных данных из набора данных болезни сердца UCI.

Визуализация конвейера выглядит следующим образом:

Пример в этой статье основан на примерах кода, содержащихся в репозитории azureml-examples . Чтобы выполнять команды локально, не копируя и вставляя YAML и другие файлы, используйте следующие команды, чтобы клонировать репозиторий и перейти в папку для языка программирования:

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

Файлы для этого примера находятся в следующих файлах:

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

Следуйте инструкциям в записных книжках Jupyter

Вы можете следовать версии пакета SDK для Python этого примера, открыв записную книжку sdk-deploy-and-test.ipynb в клонируемом репозитории.

Необходимые компоненты

Подписка Azure. Если у вас еще нет подписки Azure, создайте бесплатную учетную запись, прежде чем начинать работу.
Рабочая область Машинного обучения Azure. Сведения о создании рабочей области см. в разделе "Управление Машинное обучение Azure рабочих областей".
Следующие разрешения в рабочей области Машинное обучение Azure:
- Для создания конечных точек и развертываний пакетной службы или управления ими используйте роль владельца, участника или настраиваемую роль, назначаемую Microsoft.MachineLearningServices/workspaces/batchEndpoints/* разрешениями.
- Для создания развертываний Azure Resource Manager в группе ресурсов рабочей области: используйте владельца, участника или настраиваемую роль, Microsoft.Resources/deployments/write назначаемую разрешением в группе ресурсов, в которой развернута рабочая область.
Интерфейс командной строки Машинное обучение Azure или пакет SDK Машинное обучение Azure для Python:
- Azure CLI
- Python
Выполните следующую команду, чтобы установить Azure CLI и mlрасширение для Машинное обучение Azure:
```
az extension add -n ml
```
Развертывания компонентов конвейера для конечных точек пакетной службы представлены в версии 2.7 ml расширения для Azure CLI. az extension update --name ml Используйте команду, чтобы получить последнюю версию.
Выполните следующую команду, чтобы установить пакет SDK Машинное обучение Azure для Python:
```
pip install azure-ai-ml
```
PipelineComponentBatchDeployment Классы ModelBatchDeployment представлены в пакете SDK версии 1.7.0. pip install -U azure-ai-ml Используйте команду, чтобы получить последнюю версию.

Подключение к рабочей области

Рабочая область — это ресурс верхнего уровня для Машинного обучения Azure. Он предоставляет централизованное место для работы со всеми артефактами, создаваемыми при использовании Машинное обучение Azure. В этом разделе описано, как подключиться к рабочей области, в которой выполняются задачи развертывания.

Azure CLI
Python

В следующей команде введите идентификатор подписки, имя рабочей области, имя группы ресурсов и расположение:

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

Импортируйте необходимые библиотеки:

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

Настройте сведения о рабочей области и получите дескриптор для рабочей области:

В следующей команде введите идентификатор подписки, имя группы ресурсов и имя рабочей области:
```
subscription_id = "<subscription>"
resource_group = "<resource-group>"
workspace = "<workspace>"

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

Создание конвейера вывода

В этом разделе мы создадим все ресурсы, необходимые для нашего конвейера вывода. Начнем с создания среды, включающей необходимые библиотеки для компонентов конвейера. Затем мы создадим вычислительный кластер, в котором будет выполняться пакетное развертывание. После этого мы зарегистрируем компоненты, модели и преобразования, которые необходимо создать конвейер вывода. Наконец, мы создадим и протестируем конвейер.

Создание среды

Компоненты в этом примере будут использовать среду с XGBoost библиотеками и scikit-learn библиотеками. Файл environment/conda.yml содержит конфигурацию среды:

среда/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

Создайте среду следующим образом:

Определите среду:

Azure CLI
Python

среда/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",
)

Создайте среду:

Azure CLI
Python

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

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

Создание вычислительного кластера

Конечные точки и развертывания пакетной службы выполняются в вычислительных кластерах. Они могут работать в любом Машинное обучение Azure вычислительном кластере, который уже существует в рабочей области. Таким образом, несколько пакетных развертываний могут совместно использовать одну и ту же инфраструктуру вычислений. В этом примере мы будем работать над Машинное обучение Azure вычислительным кластеромbatch-cluster. Давайте убедимся, что вычислительные ресурсы существуют в рабочей области или создаются в противном случае.

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

Регистрация компонентов и моделей

Мы будем регистрировать компоненты, модели и преобразования, необходимые для создания конвейера вывода. Мы можем повторно использовать некоторые из этих ресурсов для учебных процедур.

Совет

В этом руководстве мы повторно будем использовать модель и компонент предварительной обработки из предыдущего конвейера обучения. Вы можете увидеть, как они были созданы, выполнив пример развертывания конвейера обучения с помощью конечных точек пакетной службы.

Зарегистрируйте модель для прогнозирования:

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

Зарегистрированная модель не была обучена непосредственно во входных данных. Вместо этого входные данные были предварительно обработаны (или преобразованы) перед обучением с помощью компонента подготовки. Нам также потребуется зарегистрировать этот компонент. Зарегистрируйте компонент подготовки:
- Azure CLI
- 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)
```
Совет

После регистрации компонента подготовки вы можете ссылаться на него из рабочей области. Например, azureml:uci_heart_prepare@latest будет получать последнюю версию компонента подготовки.
В рамках преобразований данных в компоненте подготовки входные данные нормализованы для центра прогнозировщиков и ограничения их значений в диапазоне [-1, 1]. Параметры преобразования были записаны в преобразовании scikit-learn, который также можно зарегистрировать, чтобы применить позже, когда у нас есть новые данные. Зарегистрируйте преобразование следующим образом:
- Azure CLI
- 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,
    )
)
```
Мы будем выполнять вывод для зарегистрированной модели, используя другой компонент с именем score , который вычисляет прогнозы для данной модели. Мы будем ссылать на компонент непосредственно из его определения.

Совет

Рекомендуется зарегистрировать компонент и ссылаться на него из конвейера. Однако в этом примере мы будем ссылаться на компонент непосредственно из его определения, чтобы узнать, какие компоненты повторно используются из конвейера обучения и какие новые.

Создание конвейера

Теперь пора привязать все элементы вместе. Конвейер вывода, который будет развернут, состоит из двух компонентов (шагов):

preprocess_job: этот шаг считывает входные данные и возвращает подготовленные данные и примененные преобразования. Шаг получает два входных данных:
- data: папка, содержащая входные данные для оценки
- transformations: (необязательно) Путь к преобразованиям, которые будут применены, если они доступны. При указании преобразования считываются из модели, указанной на пути. Однако если путь не указан, преобразования будут извлечены из входных данных. Однако для вывода не удается узнать параметры преобразования (в этом примере коэффициенты нормализации) из входных данных, так как необходимо использовать те же значения параметров, которые были изучены во время обучения. Так как этот вход является необязательным, preprocess_job компонент можно использовать во время обучения и оценки.
score_job: этот шаг выполнит вывод преобразованных данных с помощью входной модели. Обратите внимание, что компонент использует модель MLflow для выполнения вывода. Наконец, оценки записываются обратно в том же формате, что и они были прочитаны.

Azure CLI
Python

Конфигурация конвейера определена 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")

Давайте создадим конвейер в функции:

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

Визуализация конвейера выглядит следующим образом:

Тестирование конвейера

Давайте протестируем конвейер с некоторыми примерами данных. Для этого мы создадим задание с помощью конвейера и созданного ранее вычислительного batch-cluster кластера.

Azure CLI
Python

pipeline-job.yml Следующий файл содержит конфигурацию для задания конвейера:

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

Теперь мы настроим некоторые параметры выполнения для выполнения теста:

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

Создайте тестовое задание:

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

Создание пакетной конечной точки

Укажите имя конечной точки. Имя пакетной конечной точки должно быть уникальным в каждом регионе, так как имя используется для создания URI вызова. Чтобы обеспечить уникальность, добавьте все конечные символы в имя, указанное в следующем коде.
- Azure CLI
- Python
```
ENDPOINT_NAME="uci-classifier-score"
```
```
endpoint_name = "uci-classifier-score"
```

Настройка конечной точки:

Azure CLI
Python

Файл endpoint.yml содержит конфигурацию конечной точки.

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

Создание конечной точки:

Azure CLI
Python

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

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

Запросите универсальный код ресурса (URI) конечной точки:

Azure CLI
Python

az ml batch-endpoint show --name $ENDPOINT_NAME

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

Развертывание компонента конвейера

Чтобы развернуть компонент конвейера, необходимо создать пакетное развертывание. Развертывание — это набор ресурсов, необходимых для размещения ресурса, который выполняет фактическую работу.

Настройка развертывания
- Azure CLI
- Python
Файл deployment.yml содержит конфигурацию развертывания. Вы можете проверить схему YAML полной пакетной конечной точки для получения дополнительных свойств.

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
```
Наш конвейер определен в функции. Чтобы преобразовать его в компонент, вы будете использовать component свойство из него. Компоненты конвейера — это многократно используемые вычислительные графы, которые можно включить в пакетные развертывания или использовать для создания более сложных конвейеров.
```
pipeline_component = ml_client.components.create_or_update(
    uci_heart_classifier_scorer().component
)
```
Теперь можно определить развертывание:
```
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},
)
```
Создание развертывания
- Azure CLI
- Python
Выполните следующий код, чтобы создать пакетное развертывание в конечной точке пакетной службы и задать его как развертывание по умолчанию.
```
az ml batch-deployment create --endpoint $ENDPOINT_NAME -f deployment.yml --set-default
```
Совет

Обратите внимание на использование флага --set-default , чтобы указать, что это новое развертывание теперь используется по умолчанию.
Эта команда запустит создание развертывания и вернет ответ с подтверждением во время создания.
```
ml_client.batch_deployments.begin_create_or_update(deployment).result()
```
После создания давайте настроим это новое развертывание как стандартное:
```
endpoint = ml_client.batch_endpoints.get(endpoint_name)
endpoint.defaults.deployment_name = deployment.name
ml_client.batch_endpoints.begin_create_or_update(endpoint).result()
```
Развертывание готово для использования.

Тестирование развертывания

После создания развертывания он готов к получению заданий. Выполните следующие действия, чтобы проверить его:

Для развертывания необходимо указать один вход данных и один литеральный ввод.
- Azure CLI
- Python
Файл inputs.yml содержит определение входного ресурса данных:

inputs.yml
```
inputs:
  input_data:
    type: uri_folder
    path: data/unlabeled
  score_mode:
    type: string
    default: append
outputs:
  scores:
    type: uri_folder
    mode: upload
```
Определение ресурса входных данных:
```
input_data = Input(type="uri_folder", path="data/unlabeled/")
score_mode = Input(type="string", default="append")
```
Совет

Дополнительные сведения о том, как указывать входные данные, см. в статье "Создание заданий и входных данных" для конечных точек пакетной службы.
Вы можете вызвать развертывание по умолчанию следующим образом:
- Azure CLI
- Python
```
JOB_NAME=$(az ml batch-endpoint invoke -n $ENDPOINT_NAME --f inputs.yml --query name -o tsv)
```
Совет

Какова разница между inputs параметром и input параметром при вызове конечной точки?

Как правило, можно использовать параметр словаря inputs = {} с invoke методом для предоставления произвольного количества необходимых входных данных конечной точке пакетной службы, содержащей развертывание модели или развертывание конвейера.

Для развертывания модели можно использовать input параметр в качестве более короткого способа указать расположение входных данных для развертывания. Этот подход работает, так как развертывание модели всегда принимает только один вход данных.
```
job = ml_client.batch_endpoints.invoke(
    endpoint_name=endpoint.name,
    inputs={"input_data": input_data, "score_mode": score_mode},
)
```
Вы можете отслеживать ход выполнения отображения и потоковой передачи журналов с помощью:
- Azure CLI
- Python
```
az ml job stream -n $JOB_NAME
```
```
ml_client.jobs.get(job.name)
```
Чтобы дождаться завершения задания, выполните следующий код:
```
ml_client.jobs.stream(name=job.name)
```

Доступ к выходным данным задания

После завершения задания мы можем получить доступ к его выходным данным. Это задание содержит только один выход с именем scores:

Azure CLI
Python

Вы можете скачать связанные результаты с помощью az ml job download.

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

Скачайте результат:

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

Считывать оцененные данные:

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

Выходные данные выглядят следующим образом:

возраст	sex	...	thal	прогноз
0.9338	1	...	2	0
1.3782	1	...	3	1
1.3782	1	...	4	0
-1.954	1	...	3	0

Выходные данные содержат прогнозы, а также данные, предоставленные компоненту оценки , который был предварительно обработан. Например, столбец нормализован, а столбец agethal содержит исходные значения кодировки. На практике вы, вероятно, хотите вывести прогноз только, а затем сцепить его с исходными значениями. Эта работа была оставлена читателю.

Очистка ресурсов

После завершения удалите связанные ресурсы из рабочей области:

Azure CLI
Python

Выполните следующий код, чтобы удалить конечную точку пакетной службы и его базовое развертывание. --yes используется для подтверждения удаления.

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

Удалите конечную точку:

ml_client.batch_endpoints.begin_delete(endpoint_name)

(Необязательно) Удалите вычислительные ресурсы, если вы не планируете повторно использовать вычислительный кластер с последующими развертываниями.

Azure CLI
Python

az ml compute delete -n batch-cluster

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

Поделиться через