Использование параллельных заданий в конвейерах

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

В этой статье объясняется, как использовать интерфейс командной строки версии 2 и пакет SDK Python версии 2 для выполнения параллельных заданий в конвейерах Машинное обучение Azure. Параллельные задания ускоряют выполнение заданий путем распределения повторяющихся задач в мощных вычислительных кластерах с несколькими узлами.

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

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

Например, в сценарии, где выполняется модель обнаружения объектов в большом наборе изображений, Машинное обучение Azure параллельные задания позволяют легко распространять образы для параллельного выполнения пользовательского кода в определенном вычислительном кластере. Параллелизация может значительно сократить затраты на время. Машинное обучение Azure параллельные задания также могут упростить и автоматизировать процесс, чтобы сделать его более эффективным.

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

• У вас есть учетная запись Машинное обучение Azure и рабочая область.
• Общие сведения о конвейерах Машинное обучение Azure.

Azure CLI

• Установите Azure CLI и расширение ml. Дополнительные сведения см. в разделе Установка, настройка и использование CLI (версия 2). Расширение ml автоматически устанавливает при первом запуске az ml команды.
• Узнайте, как создавать и запускать конвейеры и компоненты Машинное обучение Azure с помощью ИНТЕРФЕЙСА командной строки версии 2.

Пакет SDK для Python

• Установите пакет SDK Машинное обучение Azure версии 2 для Python.
• Узнайте, как создавать и запускать Машинное обучение Azure конвейеры и компоненты с помощью пакета SDK для Python версии 2.

Создание и запуск конвейера с помощью параллельного шага задания

Параллельное задание Машинное обучение Azure можно использовать только как шаг в задании конвейера.

Azure CLI

Ниже приведены примеры запуска задания конвейера с помощью параллельного задания в конвейере в репозитории Машинное обучение Azure примеров.

Пакет SDK для Python

Ниже приведены примеры сборки простого конвейера машинного обучения с параллельной записной книжкой компонентов в репозитории Машинное обучение Azure примеров.

Подготовка к параллелизации

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

• Определите и привязите входные данные.
• Задайте метод деления данных.
• Настройте вычислительные ресурсы.
• Вызовите скрипт записи.

В следующих разделах описывается подготовка параллельного задания.

Объявление параметра ввода и деления данных

Для параллельного задания требуется разделить один основной вход и обрабатываться параллельно. Основной формат входных данных может быть табличным или списком файлов.

Различные форматы данных имеют разные типы входных данных, режимы ввода и методы деления данных. В следующей таблице описаны параметры:

Формат данных	Тип Ввода	Режим ввода	Метод деления данных
Список файлов	mltable или uri_folder	ro_mount или download	По размеру (количеству файлов) или по секции
Табличные данные	mltable	direct	По размеру (предполагаемому физическому размеру) или по секции

Примечание.

Если в качестве основных входных данных используются табличные mltable данные, необходимо:

• Установите библиотеку в вашей среде, как в строке mltable 9 этого файла conda.
• Укажите файл спецификации MLTable в указанном пути с заполненным разделомtransformations: - read_delimited:. Примеры см. в статье "Создание ресурсов данных и управление ими".

Вы можете объявить основные входные данные с input_data атрибутом в параллельном задании YAML или Python, а также привязать данные к определенному input параллельному заданию с помощью ${{inputs.<input name>}}. Затем вы определяете атрибут деления данных для основных входных данных в зависимости от метода деления данных.

Метод деления данных	Attribute name	Тип атрибута	Пример задания
По размеру	mini_batch_size	строка	Прогнозирование пакетной службы Iris
По секции	partition_keys	список строк	Прогнозирование продаж апельсинового сока

Настройка вычислительных ресурсов для параллелизации

После определения атрибута деления данных настройте вычислительные ресурсы для параллелизации, задав instance_count атрибуты и max_concurrency_per_instance атрибуты.

Attribute name	Type	Описание	Default value
instance_count	integer	Число узлов, используемых для задания.	1
max_concurrency_per_instance	integer	Количество процессоров на каждом узле.	Для вычислений GPU: 1. Для вычислений ЦП: количество ядер.

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

Вызов скрипта записи

Скрипт записи — это один файл Python, реализующий следующие три предопределенные функции с пользовательским кодом.

Имя функции	Обязательное поле	Description	Входные данные	Возврат
Init()	Y	Общая подготовка перед запуском мини-пакетов. Например, используйте эту функцию для загрузки модели в глобальный объект.	--	--
Run(mini_batch)	Y	Реализует основную логику выполнения для мини-пакетов.	mini_batch — это кадр данных pandas, если входные данные являются табличными данными или списком путей к файлу, если входные данные являются каталогом.	Кадры данных, список или кортеж.
Shutdown()	N	Необязательная функция для выполнения пользовательских очистки перед возвратом вычислительных ресурсов в пул.	--	--

Внимание

Чтобы избежать исключений при анализе аргументов в Init() или Run(mini_batch) функциях, используйте parse_known_args вместо parse_argsнего. См. пример iris_score скрипта записи с анализатором аргументов.

Внимание

Для Run(mini_batch) функции требуется возврат кадра данных, списка или элемента кортежа. Параллельное задание использует количество возвращаемых элементов для измерения успешности элементов в этом мини-пакете. Мини-пакетное число должно быть равно счетчику возвращаемого списка, если все элементы обработаны.

Параллельное задание выполняет функции в каждом процессоре, как показано на следующей схеме.

См. следующие примеры скриптов записи:

• Идентификация изображения для списка файлов изображений
• Классификация ирисов для табличных данных ириса

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

Attribute name	Type	Описание:
code	строка	Локальный путь к каталогу исходного кода для отправки и использования для задания.
entry_script	строка	Файл Python, содержащий реализацию предопределенных параллельных функций.

Пример шага параллельного задания

Azure CLI

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

batch_prediction:
  type: parallel
  compute: azureml:cpu-cluster
  inputs:
    input_data: 
      type: mltable
      path: ./neural-iris-mltable
      mode: direct
    score_model: 
      type: uri_folder
      path: ./iris-model
      mode: download
  outputs:
    job_output_file:
      type: uri_file
      mode: rw_mount

  input_data: ${{inputs.input_data}}
  mini_batch_size: "10kb"
  resources:
      instance_count: 2
  max_concurrency_per_instance: 2

  logging_level: "DEBUG"
  mini_batch_error_threshold: 5
  retry_settings:
    max_retries: 2
    timeout: 60

  task:
    type: run_function
    code: "./script"
    entry_script: iris_prediction.py
    environment:
      name: "prs-env"
      version: 1
      image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04
      conda_file: ./environment/environment_parallel.yml

Python

Следующий код объявляет как входные данные, привязывает job_data_path его к input_data атрибуту, задает mini_batch_size атрибут деления данных и вызывает скрипт записи.

# parallel task to process file data
file_batch_inference = parallel_run_function(
    name="file_batch_score",
    display_name="Batch Score with File Dataset",
    description="parallel component for batch score",
    inputs=dict(
        job_data_path=Input(
            type=AssetTypes.MLTABLE,
            description="The data to be split and scored in parallel",
        )
    ),
    outputs=dict(job_output_path=Output(type=AssetTypes.MLTABLE)),
    input_data="${{inputs.job_data_path}}",
    instance_count=2,
    max_concurrency_per_instance=1,
    mini_batch_size="1",
    mini_batch_error_threshold=1,
    retry_settings=dict(max_retries=2, timeout=60),
    logging_level="DEBUG",
    task=RunFunction(
        code="./src",
        entry_script="file_batch_inference.py",
        program_arguments="--job_output_path ${{outputs.job_output_path}}",
        environment="azureml://registries/azureml/environments/sklearn-1.5/labels/latest",
    ),
)

Рассмотрите параметры автоматизации

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

Ключ	Тип	Описание	Допустимые значения	Default value	Задать в аргументе атрибута или программы
mini_batch_error_threshold	integer	Количество неудачных мини-пакетов, которые будут игнорироваться в этом параллельном задании. Если число неудачных мини-пакетов выше этого порогового значения, параллельное задание помечается как неудачное. Мини-пакет помечается как неудачный, если: — количество возвращаемых данных run() меньше числа входных данных мини-пакетной службы. — исключения перехватываются в пользовательском run() коде.	[-1, int.max]	-1, что означает, что игнорировать все неудачные мини-пакеты	Атрибут mini_batch_error_threshold
mini_batch_max_retries	integer	Количество повторных попыток, когда мини-пакет завершается сбоем или истекает время ожидания. Если все повторные попытки завершаются ошибкой, мини-пакет помечается как неудачный для mini_batch_error_threshold вычисления.	[0, int.max]	2	Атрибут retry_settings.max_retries
mini_batch_timeout	integer	Время ожидания в секундах для выполнения пользовательской run() функции. Если время выполнения превышает это пороговое значение, мини-пакет прерывается и помечается как не удалось запустить повтор.	(0, 259200]	60	Атрибут retry_settings.timeout
item_error_threshold	integer	Пороговое значение неудачных элементов. Сбой элементов учитывается по числу входных данных и возвращается из каждого мини-пакета. Если сумма неудачных элементов выше этого порогового значения, параллельное задание помечается как неудачное.	[-1, int.max]	-1, что означает игнорировать все сбои во время параллельного задания	Аргумент программы --error_threshold
allowed_failed_percent	integer	mini_batch_error_thresholdАналогично, но использует процент неудачных мини-пакетов вместо количества.	[0, 100]	100	Аргумент программы --allowed_failed_percent
overhead_timeout	integer	Время ожидания в секундах для инициализации каждого мини-пакета. Например, загрузите мини-пакетные данные и передайте его в функцию run() .	(0, 259200]	600	Аргумент программы --task_overhead_timeout
progress_update_timeout	integer	Время ожидания в секундах для мониторинга хода выполнения мини-пакета. Если обновления хода выполнения не получены в течение этого параметра времени ожидания, параллельное задание помечается как неудачное.	(0, 259200]	Динамически вычисляется другими параметрами	Аргумент программы --progress_update_timeout
first_task_creation_timeout	integer	Время ожидания в секундах для мониторинга времени между запуском задания и запуском первого мини-пакета.	(0, 259200]	600	Аргумент программы --first_task_creation_timeout
logging_level	строка	Уровень журналов для дампа в файлы журналов пользователей.	INFO, WARNING или DEBUG	INFO	Атрибут logging_level
append_row_to	строка	Агрегирование всех возвращается из каждого запуска мини-пакета и выводит его в этот файл. Может ссылаться на один из выходных данных параллельного задания с помощью выражения. ${{outputs.<output_name>}}			Атрибут task.append_row_to
copy_logs_to_parent	строка	Логический параметр, следует ли копировать ход выполнения задания, обзор и журналы в родительское задание конвейера.	True или False	False	Аргумент программы --copy_logs_to_parent
resource_monitor_interval	integer	Интервал времени в секундах для дампа использования ресурсов узла (например, ЦП или памяти) для записи папки в папке logs/sys/perf . Примечание. Журналы ресурсов частого дампа немного замедляют скорость выполнения. Задайте для этого значения значение, чтобы 0 остановить использование ресурсов дампа.	[0, int.max]	600	Аргумент программы --resource_monitor_interval

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

Azure CLI

batch_prediction:
  type: parallel
  compute: azureml:cpu-cluster
  inputs:
    input_data: 
      type: mltable
      path: ./neural-iris-mltable
      mode: direct
    score_model: 
      type: uri_folder
      path: ./iris-model
      mode: download
  outputs:
    job_output_file:
      type: uri_file
      mode: rw_mount

  input_data: ${{inputs.input_data}}
  mini_batch_size: "10kb"
  resources:
      instance_count: 2
  max_concurrency_per_instance: 2

  logging_level: "DEBUG"
  mini_batch_error_threshold: 5
  retry_settings:
    max_retries: 2
    timeout: 60

  task:
    type: run_function
    code: "./script"
    entry_script: iris_prediction.py
    environment:
      name: "prs-env"
      version: 1
      image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04
      conda_file: ./environment/environment_parallel.yml
    program_arguments: >-
      --model ${{inputs.score_model}}
      --error_threshold 5
      --allowed_failed_percent 30
      --task_overhead_timeout 1200
      --progress_update_timeout 600
      --first_task_creation_timeout 600
      --copy_logs_to_parent True
      --resource_monitor_interva 20
    append_row_to: ${{outputs.job_output_file}}

Python

# parallel task to process tabular data
tabular_batch_inference = parallel_run_function(
    name="batch_score_with_tabular_input",
    display_name="Batch Score with Tabular Dataset",
    description="parallel component for batch score",
    inputs=dict(
        job_data_path=Input(
            type=AssetTypes.MLTABLE,
            description="The data to be split and scored in parallel",
        ),
        score_model=Input(
            type=AssetTypes.URI_FOLDER, description="The model for batch score."
        ),
    ),
    outputs=dict(job_output_path=Output(type=AssetTypes.MLTABLE)),
    input_data="${{inputs.job_data_path}}",
    instance_count=2,
    max_concurrency_per_instance=2,
    mini_batch_size="100",
    mini_batch_error_threshold=5,
    logging_level="DEBUG",
    retry_settings=dict(max_retries=2, timeout=60),
    task=RunFunction(
        code="./src",
        entry_script="tabular_batch_inference.py",
        environment=Environment(
            image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04",
            conda_file="./src/environment_parallel.yml",
        ),
        program_arguments="--model ${{inputs.score_model}} "
        "--job_output_path ${{outputs.job_output_path}} "
        "--error_threshold 5 "
        "--allowed_failed_percent 30 "
        "--task_overhead_timeout 1200 "
        "--progress_update_timeout 600 "
        "--first_task_creation_timeout 600 "
        "--copy_logs_to_parent True "
        "--resource_monitor_interva 20 ",
        append_row_to="${{outputs.job_output_path}}",
    ),
)

Создание конвейера с шагом параллельного задания

Azure CLI

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

$schema: https://azuremlschemas.azureedge.net/latest/pipelineJob.schema.json
type: pipeline

display_name: iris-batch-prediction-using-parallel
description: The hello world pipeline job with inline parallel job
tags:
  tag: tagvalue
  owner: sdkteam

settings:
  default_compute: azureml:cpu-cluster

jobs:
  batch_prediction:
    type: parallel
    compute: azureml:cpu-cluster
    inputs:
      input_data: 
        type: mltable
        path: ./neural-iris-mltable
        mode: direct
      score_model: 
        type: uri_folder
        path: ./iris-model
        mode: download
    outputs:
      job_output_file:
        type: uri_file
        mode: rw_mount

    input_data: ${{inputs.input_data}}
    mini_batch_size: "10kb"
    resources:
        instance_count: 2
    max_concurrency_per_instance: 2

    logging_level: "DEBUG"
    mini_batch_error_threshold: 5
    retry_settings:
      max_retries: 2
      timeout: 60

    task:
      type: run_function
      code: "./script"
      entry_script: iris_prediction.py
      environment:
        name: "prs-env"
        version: 1
        image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04
        conda_file: ./environment/environment_parallel.yml
      program_arguments: >-
        --model ${{inputs.score_model}}
        --error_threshold 5
        --allowed_failed_percent 30
        --task_overhead_timeout 1200
        --progress_update_timeout 600
        --first_task_creation_timeout 600
        --copy_logs_to_parent True
        --resource_monitor_interva 20
      append_row_to: ${{outputs.job_output_file}}

Python

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

# import required libraries
from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential
from azure.ai.ml import MLClient, Input, Output, load_component
from azure.ai.ml.dsl import pipeline
from azure.ai.ml.entities import Environment
from azure.ai.ml.constants import AssetTypes, InputOutputModes
from azure.ai.ml.parallel import parallel_run_function, RunFunction

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

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

Затем реализуйте параллельное задание, выполнив следующую процедуру parallel_run_function:

# parallel task to process tabular data
tabular_batch_inference = parallel_run_function(
    name="batch_score_with_tabular_input",
    display_name="Batch Score with Tabular Dataset",
    description="parallel component for batch score",
    inputs=dict(
        job_data_path=Input(
            type=AssetTypes.MLTABLE,
            description="The data to be split and scored in parallel",
        ),
        score_model=Input(
            type=AssetTypes.URI_FOLDER, description="The model for batch score."
        ),
    ),
    outputs=dict(job_output_path=Output(type=AssetTypes.MLTABLE)),
    input_data="${{inputs.job_data_path}}",
    instance_count=2,
    max_concurrency_per_instance=2,
    mini_batch_size="100",
    mini_batch_error_threshold=5,
    logging_level="DEBUG",
    retry_settings=dict(max_retries=2, timeout=60),
    task=RunFunction(
        code="./src",
        entry_script="tabular_batch_inference.py",
        environment=Environment(
            image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04",
            conda_file="./src/environment_parallel.yml",
        ),
        program_arguments="--model ${{inputs.score_model}} "
        "--job_output_path ${{outputs.job_output_path}} "
        "--error_threshold 5 "
        "--allowed_failed_percent 30 "
        "--task_overhead_timeout 1200 "
        "--progress_update_timeout 600 "
        "--first_task_creation_timeout 600 "
        "--copy_logs_to_parent True "
        "--resource_monitor_interva 20 ",
        append_row_to="${{outputs.job_output_path}}",
    ),
)

Наконец, используйте параллельное задание в качестве шага в конвейере и привязите его входные и выходные данные с другими шагами:

@pipeline()
def parallel_in_pipeline(pipeline_job_data_path, pipeline_score_model):

    prepare_file_tabular_data = prepare_data(input_data=pipeline_job_data_path)
    # output of file & tabular data should be type MLTable
    prepare_file_tabular_data.outputs.file_output_data.type = AssetTypes.MLTABLE
    prepare_file_tabular_data.outputs.tabular_output_data.type = AssetTypes.MLTABLE

    batch_inference_with_file_data = file_batch_inference(
        job_data_path=prepare_file_tabular_data.outputs.file_output_data
    )
    # use eval_mount mode to handle file data
    batch_inference_with_file_data.inputs.job_data_path.mode = (
        InputOutputModes.EVAL_MOUNT
    )
    batch_inference_with_file_data.outputs.job_output_path.type = AssetTypes.MLTABLE

    batch_inference_with_tabular_data = tabular_batch_inference(
        job_data_path=prepare_file_tabular_data.outputs.tabular_output_data,
        score_model=pipeline_score_model,
    )
    # use direct mode to handle tabular data
    batch_inference_with_tabular_data.inputs.job_data_path.mode = (
        InputOutputModes.DIRECT
    )

    return {
        "pipeline_job_out_file": batch_inference_with_file_data.outputs.job_output_path,
        "pipeline_job_out_tabular": batch_inference_with_tabular_data.outputs.job_output_path,
    }


pipeline_job_data_path = Input(
    path="./dataset/", type=AssetTypes.MLTABLE, mode=InputOutputModes.RO_MOUNT
)
pipeline_score_model = Input(
    path="./model/", type=AssetTypes.URI_FOLDER, mode=InputOutputModes.DOWNLOAD
)
# create a pipeline
pipeline_job = parallel_in_pipeline(
    pipeline_job_data_path=pipeline_job_data_path,
    pipeline_score_model=pipeline_score_model,
)
pipeline_job.outputs.pipeline_job_out_tabular.type = AssetTypes.URI_FILE

# set pipeline level compute
pipeline_job.settings.default_compute = "cpu-cluster"

Отправка задания конвейера

Azure CLI

Отправьте задание конвейера с параллельным шагом с помощью az ml job create команды CLI:

az ml job create --file pipeline.yml

Python

Отправка задания конвейера с параллельным шагом с помощью jobs.create_or_update функции ml_client:

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

Проверка параллельного шага в пользовательском интерфейсе Студии

После отправки задания конвейера мини-приложение SDK или CLI предоставляет веб-URL-адрес графу конвейера в пользовательском интерфейсе Студия машинного обучения Azure.

Чтобы просмотреть результаты параллельного задания, дважды щелкните параллельный шаг в графе конвейера, перейдите на вкладку "Параметры" на панели сведений, разверните узел "Выполнить параметры", а затем разверните раздел Parallel.

Чтобы отладить сбой параллельного задания, выберите вкладку "Выходные данные и журналы", разверните папку журналов и проверьте, job_result.txt, чтобы понять, почему параллельное задание завершилось сбоем. Сведения о структуре ведения журнала параллельных заданий см . в readme.txt в той же папке.

Связанный контент

• Схема параллельного задания YAML (CLI версии 2)
• Создание ресурсов данных и управление ими
• Планирование заданий конвейера машинного обучения в версии 2