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

ParallelRunConfig Класс

  • Ссылка

Определяет конфигурацию для объекта ParallelRunStep.

Пример использования ParallelRunStep приведен в записной книжке: https://aka.ms/batch-inference-notebooks.

Руководство по устранению неполадок приведено в https://aka.ms/prstsg. Там можно найти дополнительные ссылки.

Инициализируйте объект конфигурации.

Наследование
azureml.pipeline.core._parallel_run_config_base._ParallelRunConfigBase
ParallelRunConfig

Конструктор

ParallelRunConfig(environment, entry_script, error_threshold, output_action, compute_target, node_count, process_count_per_node=None, mini_batch_size=None, source_directory=None, description=None, logging_level='INFO', run_invocation_timeout=60, run_max_try=3, append_row_file_name=None, allowed_failed_count=None, allowed_failed_percent=None, partition_keys=None, environment_variables=None)

Параметры

Имя Описание
environment
Обязательно
Environment

Определение среды, с помощью которого происходит настройка среды Python. Настройку можно выполнить для использования существующей среды Python или временной среды для указанного эксперимента. Определение среды используется для определения необходимых зависимостей приложений, таких как пакеты Conda или PIP.
entry_script
Обязательно
str

Пользовательский скрипт, который будет выполняться параллельно на нескольких узлах. Это значение указывается в качестве пути к локальному файлу. Если указан source_directory, то entry_script является относительным путем внутри каталога. В противном случае это может быть любой путь, доступный на компьютере. entry_script должен содержать две функции: init() — эта функция должна использоваться для любой дорогостоящей или распространенной подготовки к последующим выводам, например, десериализации и загрузки модели в глобальный объект. run(mini_batch): Метод для параллелизации. Каждый вызов будет иметь один мини-пакет. mini_batch: пакетный вывод вызывает метод run и передает ему в качестве аргумента список либо кадр данных Pandas. Каждая запись в min_batch содержит одно из следующих значений: путь к файлу для входных данных в формате FileDataset и Pandas DataFrame для входных данных в формате TabularDataset. Метод run() должен возвращать Pandas DataFrame или массив. Для append_row output_action эти возвращаемые элементы добавляются в общий выходной файл. Для summary_only содержимое элементов игнорируется. Для всех выходных действий каждый возвращаемый элемент обозначает один успешный вывод для входного элемента во входном мини-пакете. Каждый параллельный рабочий процесс будет вызывать метод init один раз, а затем выполнять циклический проход функции run до тех пор, пока не будут обработаны все мини-пакеты.
error_threshold
Обязательно
int

Количество ошибок записи для TabularDataset и сбоев чтения файлов для FileDataset, которые следует игнорировать во время обработки. Если число ошибок превышает это значение, задание будет прервано. Пороговое значение количества ошибок применяется к общему объему входных данных, а не к отдельным мини-пакетам, которые передаются в метод run(). Диапазон значений — [-1, int. max]. -1 указывает на игнорирование всех ошибок во время обработки.
output_action
Обязательно
str

Способ упорядочения выходных данных. Текущие поддерживаемые значения: append_row и summary_only.

  1. append_row: все значения, выводимые вызовами метода run(), будут объединены в один уникальный файл с именем parallel_run_step.txt, который создается в расположении выходных данных.
  2. summary_only: требуется пользовательский скрипт для хранения выходных данных. Для каждого успешно обработанного входного элемента по-прежнему ожидается выходная запись. Система использует эти выходные данные только для вычисления порогового значения ошибок (игнорируя фактическое значение записи).
compute_target
Обязательно
AmlCompute или str

Целевой объект вычислений, используемый для выполнения ParallelRunStep. Этот параметр можно указать как целевой объект вычислений или имя целевого объекта вычислений в рабочей области.
node_count
Обязательно
int

Количество узлов в целевом объекте вычислений, используемых для выполнения ParallelRunStep.
process_count_per_node
int

Количество рабочих процессов на узел для параллельного выполнения начального сценария. Для компьютера с GPU значение по умолчанию равно 1. Для компьютера с графическим ЦП значение по умолчанию равно количеству ядер. Рабочий процесс будет многократно вызывать run(), передавая полученный мини-пакет. Общее число рабочих процессов в задании равно process_count_per_node * node_count, что определяет максимальное число run() для параллельного выполнения.

Default value: None
mini_batch_size
Union[str, int]

Для входных данных FileDataset значением этого поля является число файлов, которые может обработать пользовательский скрипт в одном вызове метода run(). Для входных данных TabularDataset значение этого поля представляет приблизительный размер данных, которые может обработать пользовательский скрипт в одном вызове метода run(). Примеры значений: 1024, 1024 КБ, 10 МБ и 1 ГБ. (Необязательно. Значение по умолчанию — 10 файлов для FileDataset и 1 МБ для TabularDataset.)

Default value: None
source_directory
str

Путь к папке, содержащей entry_script и вспомогательные файлы, используемые для выполнения в целевом объекте вычислений.

Default value: None
description
str

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

Default value: None
logging_level
str

Строка имени уровня ведения журнала, которая определена в параметре "logging". Возможные значения: "WARNING" (Предупреждение), "INFO" (Информация) и "DEBUG" (Отладка). (Необязательно. Значение по умолчанию — "INFO".)

Default value: INFO
run_invocation_timeout
int

Время ожидания в секундах для каждого вызова метода run(). (Необязательно. Значение по умолчанию — 60.)

Default value: 60
run_max_try
int

Максимальное количество попыток для мини-пакета с ошибкой или истечением времени ожидания. Диапазон значений — [1, int.max]. Значение по умолчанию равно 3. Мини-пакет с количеством выводов из очереди, превышающим это значение, не будет обрабатываться повторно и будет удален напрямую.

Default value: 3
append_row_file_name
str

Имя выходного файла, если output_action имеет значение "append_row". (Необязательно. Значение по умолчанию — "parallel_run_step.txt".)

Default value: None
allowed_failed_count
int

Количество мини-пакетов с ошибкой, которые не должны учитываться во время обработки. Если число мини-пакетов с ошибкой превышает это значение, задание будет прервано. Это пороговое значение предназначено для всех входных данных, а не для отдельного мини-пакета, отправленного в метод run(). Диапазон значений — [-1, int. max]. -1 указывает на игнорирование всех ошибок во время обработки. Мини-пакет может завершиться сбоем при первой обработке, а затем успешно пройти обработку во второй раз. Проверка между первым и вторым временем будет считаться ошибкой. Проверка после второго времени не будет считаться ошибкой. Аргументы –error_threshold, –allowed_failed_count и –allowed_failed_percent могут работать вместе. Если указано несколько аргументов, задание будет прервано при превышении заданного значения.

Default value: None
allowed_failed_percent
float

Процент мини-пакетов с ошибкой, которые не должны учитываться во время обработки. Если процент мини-пакетов с ошибкой превышает это значение, задание будет прервано. Это пороговое значение предназначено для всех входных данных, а не для отдельного мини-пакета, отправленного в метод run(). Диапазон — [0, 100]. 100 или 100.0 указывает на игнорирование всех ошибок во время обработки. Проверка начнется после планирования выполнения всех мини-пакетов. Аргументы –error_threshold, –allowed_failed_count и –allowed_failed_percent могут работать вместе. Если указано несколько аргументов, задание будет прервано при превышении заданного значения.

Default value: None
partition_keys
List[str]

Ключи, используемые для секционирования набора данных на мини-пакеты. Если этот параметр указан, данные с одним и тем же ключом будут секционированы в один мини-пакет. Если указаны и partition_keys, и mini_batch_size, возникнет ошибка. Это должен быть список элементов str, каждый из которых является ключом, используемым для секционирования входного набора данных. Однако при передаче в PipelineParameter значения по умолчанию должны быть строками дампа JSON списка, так как тип списка в PipelineParameter пока не поддерживается. Входные данные должны быть секционированными наборами данных, а partition_keys должен представлять собой подмножество ключей каждого входного набора.

Default value: None
environment_variables
Dict[str, str]

Словарь имен и значений переменных среды. Эти переменные среды задаются для процесса, в котором выполняется пользовательский скрипт.

Default value: None
environment
Обязательно
Environment

Определение среды, с помощью которого происходит настройка среды Python. Настройку можно выполнить для использования существующей среды Python или временной среды для указанного эксперимента. Определение среды используется для определения необходимых зависимостей приложений, таких как пакеты Conda или PIP.
entry_script
Обязательно
str

Пользовательский скрипт, который будет выполняться параллельно на нескольких узлах. Это значение указывается в качестве пути к локальному файлу. Если указан source_directory, то entry_script является относительным путем внутри каталога. В противном случае это может быть любой путь, доступный на компьютере. entry_script должен содержать две функции: init() — эта функция должна использоваться для любой дорогостоящей или распространенной подготовки к последующим выводам, например, десериализации и загрузки модели в глобальный объект. run(mini_batch): Метод для параллелизации. Каждый вызов будет иметь один мини-пакет. mini_batch: пакетный вывод вызывает метод run и передает ему в качестве аргумента список либо кадр данных Pandas. Каждая запись в min_batch содержит одно из следующих значений: путь к файлу для входных данных в формате FileDataset и Pandas DataFrame для входных данных в формате TabularDataset. Метод run() должен возвращать Pandas DataFrame или массив. Для append_row output_action эти возвращаемые элементы добавляются в общий выходной файл. Для summary_only содержимое элементов игнорируется. Для всех выходных действий каждый возвращаемый элемент обозначает один успешный вывод для входного элемента во входном мини-пакете. Каждый параллельный рабочий процесс будет вызывать метод init один раз, а затем выполнять циклический проход функции run до тех пор, пока не будут обработаны все мини-пакеты.
error_threshold
Обязательно
int

Количество ошибок записи для TabularDataset и сбоев чтения файлов для FileDataset, которые следует игнорировать во время обработки. Если число ошибок превышает это значение, задание будет прервано. Пороговое значение количества ошибок применяется к общему объему входных данных, а не к отдельным мини-пакетам, которые передаются в метод run(). Диапазон значений — [-1, int. max]. -1 указывает на игнорирование всех ошибок во время обработки.
output_action
Обязательно
str

Способ упорядочения выходных данных. Текущие поддерживаемые значения: append_row и summary_only.

  1. append_row: все значения, выводимые вызовами метода run(), будут объединены в один уникальный файл с именем parallel_run_step.txt, который создается в расположении выходных данных.
  2. summary_only: требуется пользовательский скрипт для хранения выходных данных. Для каждого успешно обработанного входного элемента по-прежнему ожидается выходная запись. Система использует эти выходные данные только для вычисления порогового значения ошибок (игнорируя фактическое значение записи).
compute_target
Обязательно
AmlCompute или str

Целевой объект вычислений, используемый для выполнения ParallelRunStep. Этот параметр можно указать как целевой объект вычислений или имя целевого объекта вычислений в рабочей области.
node_count
Обязательно
int

Количество узлов в целевом объекте вычислений, используемых для выполнения ParallelRunStep.
process_count_per_node
Обязательно
int

Количество рабочих процессов на узел для параллельного выполнения начального сценария. Для компьютера GPU значение по умолчанию равно 1. Для компьютера ЦП значением по умолчанию является количество ядер. Рабочий процесс будет многократно вызывать run(), передавая полученный мини-пакет. Общее число рабочих процессов в задании равно process_count_per_node * node_count, что определяет максимальное число run() для параллельного выполнения.
mini_batch_size
Обязательно
str или int

Для входных данных FileDataset значением этого поля является число файлов, которые может обработать пользовательский скрипт в одном вызове метода run(). Для входных данных TabularDataset значение этого поля представляет приблизительный размер данных, которые может обработать пользовательский скрипт в одном вызове метода run(). Примеры значений: 1024, 1024 КБ, 10 МБ и 1 ГБ. (Необязательно. Значение по умолчанию — 10 файлов для FileDataset и 1 МБ для TabularDataset.)
source_directory
Обязательно
str

Путь к папке, содержащей entry_script и вспомогательные файлы, используемые для выполнения в целевом объекте вычислений.
description
Обязательно
str

Описание пакетной службы, используемой в целях отображения.
logging_level
Обязательно
str

Строка имени уровня ведения журнала, которая определена в параметре "logging". Возможные значения: "WARNING" (Предупреждение), "INFO" (Информация) и "DEBUG" (Отладка). (Необязательно. Значение по умолчанию — "INFO".)
run_invocation_timeout
Обязательно
int

Время ожидания в секундах для каждого вызова метода run(). (Необязательно. Значение по умолчанию — 60.)
run_max_try
Обязательно
int

Максимальное количество попыток для мини-пакета с ошибкой или истечением времени ожидания. Диапазон значений — [1, int.max]. Значение по умолчанию равно 3. Мини-пакет с количеством выводов из очереди, превышающим это значение, не будет обрабатываться повторно и будет удален напрямую.
append_row_file_name
Обязательно
str

Имя выходного файла, если output_action имеет значение "append_row". (Необязательно. Значение по умолчанию — "parallel_run_step.txt".)
allowed_failed_count
Обязательно
int

Количество мини-пакетов с ошибкой, которые не должны учитываться во время обработки. Если число мини-пакетов с ошибкой превышает это значение, задание будет прервано. Это пороговое значение предназначено для всех входных данных, а не для отдельного мини-пакета, отправленного в метод run(). Диапазон значений — [-1, int. max]. -1 указывает на игнорирование всех ошибок во время обработки. Мини-пакет может завершиться сбоем при первой обработке, а затем успешно пройти обработку во второй раз. Проверка между первым и вторым временем будет считаться ошибкой. Проверка после второго времени не будет считаться ошибкой. Аргументы –error_threshold, –allowed_failed_count и –allowed_failed_percent могут работать вместе. Если указано несколько аргументов, задание будет прервано при превышении заданного значения.
allowed_failed_percent
Обязательно
float

Процент мини-пакетов с ошибкой, которые не должны учитываться во время обработки. Если процент мини-пакетов с ошибкой превышает это значение, задание будет прервано. Это пороговое значение предназначено для всех входных данных, а не для отдельного мини-пакета, отправленного в метод run(). Диапазон — [0, 100]. 100 или 100.0 указывает на игнорирование всех ошибок во время обработки. Проверка начнется после планирования выполнения всех мини-пакетов. Аргументы –error_threshold, –allowed_failed_count и –allowed_failed_percent могут работать вместе. Если указано несколько аргументов, задание будет прервано при превышении заданного значения.
partition_keys
Обязательно
List[str]

Ключи, используемые для секционирования набора данных на мини-пакеты. Если этот параметр указан, данные с одним и тем же ключом будут секционированы в один мини-пакет. Если указаны и partition_keys, и mini_batch_size, возникнет ошибка. Это должен быть список элементов str, каждый из которых является ключом, используемым для секционирования входного набора данных. Однако при передаче в PipelineParameter значения по умолчанию должны быть строками дампа JSON списка, так как тип списка в PipelineParameter пока не поддерживается. Входные данные должны быть секционированными наборами данных, а partition_keys должен представлять собой подмножество ключей каждого входного набора.
environment_variables
Обязательно
Dict[str, str]

Словарь имен и значений переменных среды. Эти переменные среды задаются для процесса, в котором выполняется пользовательский скрипт.

Комментарии

Класс ParallelRunConfig используется для предоставления конфигурации для класса ParallelRunStep. ParallelRunConfig и ParallelRunStep можно использовать вместе для параллельной обработки больших объемов данных. Распространенные варианты использования — это обучение модели ML (Машинного обучения) или запуск автономного вывода для создания прогнозов на основе пакета наблюдений. ParallelRunStep работает, разбивая данные на пакеты, которые обрабатываются параллельно. Размер пакета, количество узлов и другие настраиваемые параметры для ускорения параллельной обработки можно контролировать с помощью класса ParallelRunConfig. ParallelRunStep может работать с TabularDataset или FileDataset в качестве входных данных.

Чтобы использовать ParallelRunStep и ParallelRunConfig, выполните приведенные ниже действия:

  • Создайте объект ParallelRunConfig, чтобы указать, как выполняется пакетная обработка, с параметрами для управления размером пакета, количеством узлов на целевой объект вычислений и ссылкой на пользовательский скрипт Python.

  • Создайте объект ParallelRunStep, который использует объект ParallelRunConfig, определяет входные и выходные данные для этапа.

  • Используйте настроенный объект ParallelRunStep в Pipeline так же, как это делается для других типов шагов конвейера.

Примеры работы с классами ParallelRunStep и ParallelRunConfig для вывода пакетов рассматриваются в следующих статьях:


   from azureml.pipeline.steps import ParallelRunStep, ParallelRunConfig

   parallel_run_config = ParallelRunConfig(
       source_directory=scripts_folder,
       entry_script=script_file,
       mini_batch_size="5",        # or partition_keys=["key1", "key2"], which is another way to partition the
                                   # input to mini-batches, refer to the parameter description for details
       error_threshold=10,         # Optional, allowed failed count on mini batch items
       allowed_failed_count=15,    # Optional, allowed failed count on mini batches
       allowed_failed_percent=10,  # Optional, allowed failed percent on mini batches
       run_max_try=3,
       output_action="append_row",
       environment=batch_env,
       compute_target=compute_target,
       node_count=2)

   parallelrun_step = ParallelRunStep(
       name="predict-digits-mnist",
       parallel_run_config=parallel_run_config,
       inputs=[ named_mnist_ds ],
       output=output_dir,
       arguments=[ "--extra_arg", "example_value" ],
       allow_reuse=True
   )

Дополнительные сведения об этом примере см. в записной книжке https://aka.ms/batch-inference-notebooks.

Методы

load_yaml

Загрузка данных конфигурации параллельного выполнения из файла YAML.
save_to_yaml

Экспорт данных конфигурации параллельного выполнения в файл YAML.

load_yaml

Загрузка данных конфигурации параллельного выполнения из файла YAML.

static load_yaml(workspace, path)

Параметры

Имя Описание
workspace
Обязательно
Workspace

Рабочая область, из которой считываются данные конфигурации.
path
Обязательно
str

Путь, из которого загружается конфигурация.

save_to_yaml

Экспорт данных конфигурации параллельного выполнения в файл YAML.

save_to_yaml(path)

Параметры

Имя Описание
path
Обязательно
str

Путь для сохраняемого файла.