Активация конвейеров машинного обучения

Статья
06/03/2023

ОБЛАСТЬ ПРИМЕНЕНИЯ:Пакет SDK для Python azureml версии 1

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

Предварительные требования

Подписка Azure. Если у вас еще нет подписки Azure, создайте бесплатную учетную запись.
Среда Python, в которой установлен пакет Машинного обучения Azure SDK для Python. Дополнительные сведения см. в статьях Создание повторно используемых сред для обучения и развертывания с помощью Машинного обучения Azure и управление ими.
Рабочая область Машинного обучения с опубликованным конвейером. Можно использовать рабочую область из раздела Машинное обучение Azure SDK.

Активация конвейеров с помощью пакета Машинного обучения Azure SDK для Python

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

import azureml.core
from azureml.core import Workspace
from azureml.pipeline.core import Pipeline, PublishedPipeline
from azureml.core.experiment import Experiment

ws = Workspace.from_config()

experiments = Experiment.list(ws)
for experiment in experiments:
    print(experiment.name)

published_pipelines = PublishedPipeline.list(ws)
for published_pipeline in  published_pipelines:
    print(f"{published_pipeline.name},'{published_pipeline.id}'")

experiment_name = "MyExperiment" 
pipeline_id = "aaaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"

Создание расписания

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

В верхней части файла Python импортируйте классы Schedule и ScheduleRecurrence:


from azureml.pipeline.core.schedule import ScheduleRecurrence, Schedule

Создание расписания на основе времени

Конструктор ScheduleRecurrence имеет обязательный аргумент frequency в виде одной из следующих строк: "Minute", "Hour", "Day", "Week" или "Month". Кроме того, требуется целочисленный аргумент interval, указывающий, сколько единиц frequency должно пройти между запусками расписания. Необязательные аргументы позволяют подробнее указывать время запуска, как описано в документации по пакету ScheduleRecurrence SDK.

Создайте Schedule, который будет начинать выполнение задания каждые 15 минут:

recurrence = ScheduleRecurrence(frequency="Minute", interval=15)
recurring_schedule = Schedule.create(ws, name="MyRecurringSchedule", 
                            description="Based on time",
                            pipeline_id=pipeline_id, 
                            experiment_name=experiment_name, 
                            recurrence=recurrence)

Создание расписания на основе изменений

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

Примечание

Расписания на основе изменений поддерживают только мониторинг хранилища BLOB-объектов Azure.

Для создания реактивного файла Schedule необходимо задать параметр datastore в вызове Schedule.create. Для мониторинга папки задайте аргумент path_on_datastore.

Аргумент polling_interval позволяет указать в минутах частоту проверки хранилища данных на наличие изменений.

Если конвейер создан с применением DataPath PipelineParameter, можно присвоить этой переменной имя измененного файла, задав аргумент data_path_parameter_name.

datastore = Datastore(workspace=ws, name="workspaceblobstore")

reactive_schedule = Schedule.create(ws, name="MyReactiveSchedule", description="Based on input file change.",
                            pipeline_id=pipeline_id, experiment_name=experiment_name, datastore=datastore, data_path_parameter_name="input_data")

Дополнительные аргументы при создании расписания

В дополнение к описанным выше аргументам, можно задать для аргумента status значение "Disabled", чтобы сформировать неактивное расписание. Наконец, continue_on_step_failure позволяет передавать логическое значение для переопределения поведения по умолчанию при сбоях конвейера.

Просмотр запланированных конвейеров

В веб-браузере перейдите в раздел "Машинное обучение Azure". В разделе Конечные точки панели навигации выберите пункт Конечные точки конвейера. Откроется список конвейеров, опубликованных в рабочей области.

Страница конвейеров AML

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

Деактивация конвейера

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

pipeline = PublishedPipeline.get(ws, id=pipeline_id)
pipeline.disable()

Если конвейер запланирован, сначала необходимо отменить его запуск. Получите идентификатор расписания на портале или следующей командой.

ss = Schedule.list(ws)
for s in ss:
    print(s)

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

def stop_by_schedule_id(ws, schedule_id):
    s = next(s for s in Schedule.list(ws) if s.id == schedule_id)
    s.disable()
    return s

stop_by_schedule_id(ws, schedule_id)

При повторном запуске Schedule.list(ws) вы получите пустой список.

Использование Azure Logic Apps для сложных активаторов

Более сложные правила активатора или поведения можно создавать с помощью Azure Logic App.

Чтобы использовать Azure Logic App для активации конвейера Машинного обучения, требуется конечная точка REST для опубликованного конвейера Машинного обучения. Создайте и опубликуйте свой конвейер. Затем найдите конечную точку REST PublishedPipeline с помощью идентификатора конвейера.

# You can find the pipeline ID in Azure Machine Learning studio

published_pipeline = PublishedPipeline.get(ws, id="<pipeline-id-here>")
published_pipeline.endpoint

Создание приложения логики в Azure

Теперь создайте экземпляр Azure Logic App. После подготовки приложения логики выполните следующие действия, чтобы настроить триггер для конвейера:

Чтобы предоставить приложению доступ к рабочей области Машинного обучения Azure, создайте назначаемое системой управляемое удостоверение.
Перейдите в представление конструктора приложений Azure Logic App и выберите пустой шаблон приложения Azure Logic App.
В Конструкторе найдите BLOB-объект. Выберите активатор При добавлении или изменении BLOB-объект (только свойства) и добавьте его в Azure Logic App.
Укажите сведения о подключении для учетной записи хранилища BLOB-объектов, которую вы хотите отслеживать для добавления или изменения BLOB-объектов. Выберите контейнер для отслеживания.

Выберите интервал и частоту опроса обновлений, которые для вас работают.

Примечание

Этот активатор будет отслеживать выбранный контейнер, но не вложенные папки.
Добавьте действие HTTP, которое будет выполняться при обнаружении нового или измененного BLOB-объекта. Выберите + Новый шаг, затем найдите и выберите действие HTTP.

Чтобы настроить действие, используйте следующие параметры.

Параметр	Значение
Действие HTTP	POST
URI	Конечная точка для опубликованного конвейера, которая находится как предварительный компонент
Режим проверки подлинности	Управляемое удостоверение

Настройте расписание для задания любого пути к данным PipelineParameters, который может быть в наличии.
```
{
  "DataPathAssignments": {
    "input_datapath": {
      "DataStoreName": "<datastore-name>",
      "RelativePath": "@{triggerBody()?['Name']}" 
    }
  },
  "ExperimentName": "MyRestPipeline",
  "ParameterAssignments": {
    "input_string": "sample_string3"
  },
  "RunSource": "SDK"
}
```
Используйте DataStoreName, добавленный в рабочую область в качестве предварительного компонента.
Выберите Сохранить — ваше расписание теперь готово.

Важно!

При работе с доступом на основе ролей Azure (Azure RBAC) для управления доступом к конвейеру задайте разрешения для своего сценария конвейера (обучение или оценка).

Вызов конвейеров машинного обучения из конвейеров Фабрики данных Azure

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

Снимок экрана, показывающий действие конвейера Машинного обучения в среде разработки Фабрики данных Azure

Дальнейшие действия

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

Дополнительные сведения см. в разделе:

Использование конвейеров Машинного обучения Azure для пакетной оценки

Дополнительные сведения о конвейерах
Дополнительные сведения об исследовании Машинного обучения Azure с помощью Jupyter