руководство: запуск параллельной рабочей нагрузки с Azure Batch с помощью API Python

Используйте пакетную службу Azure, чтобы эффективно выполнять пакетные задания для крупномасштабных параллельных и высокопроизводительных вычислений (HPC). В этом руководстве рассматривается пример выполнения параллельной рабочей нагрузки на Python с использованием сервиса Azure Batch. Вы изучите общий рабочий процесс приложения пакетной службы и узнаете, как программно взаимодействовать с ресурсами пакетной службы и службы хранилища.

В этом руководстве вы преобразуете файлы мультимедиа MP4 в формат MP3 параллельно с помощью средства ffmpeg с открытым исходным кодом.

Если у вас нет аккаунта Azure, создайте бесплатную учетную запись перед началом.

Предпосылки

• Python версии 3.8 или более поздней

• Диспетчер пакетов pip

• Учетная запись пакетной службы Azure и связанная учетная запись хранения Azure. Чтобы создать эти учетные записи, ознакомьтесь с краткими руководствами по Azure Batch для портала Azure или Azure CLI.

Вход в Azure

Войдите на портал Azure.

Получение учетных данных учетной записи

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

1. Выберите Все службы>Пакетные учетные записи и затем выберите имя вашей пакетной учетной записи.

2. Для просмотра учетных данных пакетной обработки выберите Ключи. Скопируйте значения учетной записи пакетной службы, URL-адреса, и первичного ключа доступа в текстовый редактор.

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

Загрузка и запуск примера приложения

Скачивание примера приложения

Скачайте или клонируйте пример приложения с GitHub. Чтобы клонировать пример репозитория приложения с клиентом Git, выполните следующую команду:

git clone https://github.com/Azure-Samples/batch-python-ffmpeg-tutorial.git

Перейдите в каталог, содержащий файл batch_python_tutorial_ffmpeg.py.

В среде Python установите необходимые пакеты с помощью pip.

pip install -r requirements.txt

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

_BATCH_ACCOUNT_NAME = 'yourbatchaccount'
_BATCH_ACCOUNT_KEY = 'xxxxxxxxxxxxxxxxE+yXrRvJAqT9BlXwwo1CwF+SwAYOxxxxxxxxxxxxxxxx43pXi/gdiATkvbpLRl3x14pcEQ=='
_BATCH_ACCOUNT_URL = 'https://yourbatchaccount.yourbatchregion.batch.azure.com'
_STORAGE_ACCOUNT_NAME = 'mystorageaccount'
_STORAGE_ACCOUNT_KEY = 'xxxxxxxxxxxxxxxxy4/xxxxxxxxxxxxxxxxfwpbIC5aAWA8wDu+AFXZB827Mt9lybZB1nUcQbQiUrkPtilK5BQ=='

Запуск приложения

Выполните следующее, чтобы запустить этот сценарий.

python batch_python_tutorial_ffmpeg.py

Когда вы запускаете образец приложения, вывод в консоли будет примерно следующим. Во время выполнения вы испытываете паузу на этапе Monitoring all tasks for 'Completed' state, timeout in 00:30:00..., когда запускаются вычислительные узлы пула.

Sample start: 11/28/2018 3:20:21 PM

Container [input] created.
Container [output] created.
Uploading file LowPriVMs-1.mp4 to container [input]...
Uploading file LowPriVMs-2.mp4 to container [input]...
Uploading file LowPriVMs-3.mp4 to container [input]...
Uploading file LowPriVMs-4.mp4 to container [input]...
Uploading file LowPriVMs-5.mp4 to container [input]...
Creating pool [LinuxFFmpegPool]...
Creating job [LinuxFFmpegJob]...
Adding 5 tasks to job [LinuxFFmpegJob]...
Monitoring all tasks for 'Completed' state, timeout in 00:30:00...
Success! All tasks completed successfully within the specified timeout period.
Deleting container [input]....

Sample end: 11/28/2018 3:29:36 PM
Elapsed time: 00:09:14.3418742

Перейдите в учетную запись Batch на портале Azure, чтобы контролировать пул, вычислительные узлы, задания и задачи. Например, чтобы просмотреть тепловую диаграмму вычислительных узлов в пуле, выберите Пулы>LinuxFFmpegPool.

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

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

Извлечение выходных файлов

Портал Azure можно использовать для скачивания выходных MP3-файлов, созданных задачами ffmpeg.

1. Выберите Все службы>Учетные записи службы хранилища и щелкните имя учетной записи службы хранилища.
2. Щелкните Blobs>вывод.
3. Щелкните правой кнопкой мыши один из выходных MP3-файлов и нажмите кнопку Загрузить. Следуйте инструкциям в браузере, чтобы открыть или сохранить этот файл.

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

Просмотр кода

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

Аутентификация Blob и Batch клиентов

Для взаимодействия с учетной записью хранения приложение использует пакет azure-storage-blob для создания объекта BlockBlobService .

blob_client = azureblob.BlockBlobService(
    account_name=_STORAGE_ACCOUNT_NAME,
    account_key=_STORAGE_ACCOUNT_KEY)

Приложение создает объект BatchServiceClient для создания пулов, заданий и задач в пакетной службе и управления ими. Клиент пакетной службы в примере использует проверку подлинности с общим ключом. Batch также поддерживает проверку подлинности с помощью Microsoft Entra ID, чтобы аутентифицировать отдельных пользователей или автоматическое приложение.

credentials = batchauth.SharedKeyCredentials(_BATCH_ACCOUNT_NAME,
                                             _BATCH_ACCOUNT_KEY)

batch_client = batch.BatchServiceClient(
    credentials,
    base_url=_BATCH_ACCOUNT_URL)

Передача входных файлов

Приложение использует ссылку blob_client для создания контейнера хранилища для входных MP4-файлов и контейнера для выходных данных задачи. Затем вызывается функция upload_file_to_container, чтобы отправить MP4-файлы из локальной директории InputFiles в контейнер. Файлы в хранилище определяются как объекты Batch ResourceFile, которые Batch может впоследствии загрузить на вычислительные узлы.

blob_client.create_container(input_container_name, fail_on_exist=False)
blob_client.create_container(output_container_name, fail_on_exist=False)
input_file_paths = []

for folder, subs, files in os.walk(os.path.join(sys.path[0], './InputFiles/')):
    for filename in files:
        if filename.endswith(".mp4"):
            input_file_paths.append(os.path.abspath(
                os.path.join(folder, filename)))

# Upload the input files. This is the collection of files that are to be processed by the tasks.
input_files = [
    upload_file_to_container(blob_client, input_container_name, file_path)
    for file_path in input_file_paths]

Создание пула вычислительных узлов

Затем в учетной записи пакетной службы создается пул вычислительных узлов с помощью вызова create_pool. Эта определяемая функция использует класс Batch PoolAddParameter для задания количества узлов, размера виртуальной машины и конфигурации пула. Здесь объект VirtualMachineConfiguration указывает ImageReference для образа Ubuntu Server 20.04 LTS, опубликованного в Azure Marketplace. Служба Azure Batch поддерживает широкий спектр образов ВМ в торговой площадке Azure Marketplace и пользовательских образов ВМ.

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

Помимо свойств физического узла, эта конфигурация пула включает объект StartTask . StartTask выполняется на каждом узле, когда этот узел присоединяется к пулу, и каждый раз, когда узел перезапускается. В этом примере StartTask запускает команды оболочки Bash для установки пакета ffmpeg и зависимостей на узлах.

Метод pool.add отправляет пул в пакетную службу.

new_pool = batch.models.PoolAddParameter(
    id=pool_id,
    virtual_machine_configuration=batchmodels.VirtualMachineConfiguration(
        image_reference=batchmodels.ImageReference(
            publisher="Canonical",
            offer="UbuntuServer",
            sku="20.04-LTS",
            version="latest"
        ),
        node_agent_sku_id="batch.node.ubuntu 20.04"),
    vm_size=_POOL_VM_SIZE,
    target_dedicated_nodes=_DEDICATED_POOL_NODE_COUNT,
    target_low_priority_nodes=_LOW_PRIORITY_POOL_NODE_COUNT,
    start_task=batchmodels.StartTask(
        command_line="/bin/bash -c \"apt-get update && apt-get install -y ffmpeg\"",
        wait_for_success=True,
        user_identity=batchmodels.UserIdentity(
            auto_user=batchmodels.AutoUserSpecification(
                scope=batchmodels.AutoUserScope.pool,
                elevation_level=batchmodels.ElevationLevel.admin)),
    )
)
batch_service_client.pool.add(new_pool)

Создание задания

Пакетное задание указывает пул для запуска задач и дополнительные параметры, такие как приоритет и расписание работы. Пример создает задание путем вызова create_job. Данная функция использует класс JobAddParameter для создания задания в вашем пуле. Метод job.add отправляет пул в пакетную службу. Изначально задание не имеет задач.

job = batch.models.JobAddParameter(
    id=job_id,
    pool_info=batch.models.PoolInformation(pool_id=pool_id))

batch_service_client.job.add(job)

Создание задач

Приложение создает задачи в рабочем процессе посредством вызова add_tasks. Эта определяемая функция создает список объектов задач с помощью класса TaskAddParameter . Каждая задача запускает ffmpeg для обработки входного resource_files объекта с помощью command_line параметра. ffmpeg был ранее установлен на каждом узле при создании пула. В командной строке выполняется ffmpeg для преобразования каждого входного файла MP4 (видео) в файл MP3 (аудио).

В примере создается объект OutputFile для файла MP3 после запуска командной строки. Выходные файлы каждой задачи (в данном случае — один) загружаются в контейнер связанной учетной записи хранения с использованием свойства задачи output_files.

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

tasks = list()

for idx, input_file in enumerate(input_files):
    input_file_path = input_file.file_path
    output_file_path = "".join((input_file_path).split('.')[:-1]) + '.mp3'
    command = "/bin/bash -c \"ffmpeg -i {} {} \"".format(
        input_file_path, output_file_path)
    tasks.append(batch.models.TaskAddParameter(
        id='Task{}'.format(idx),
        command_line=command,
        resource_files=[input_file],
        output_files=[batchmodels.OutputFile(
            file_pattern=output_file_path,
            destination=batchmodels.OutputFileDestination(
                container=batchmodels.OutputFileBlobContainerDestination(
                    container_url=output_container_sas_url)),
            upload_options=batchmodels.OutputFileUploadOptions(
                upload_condition=batchmodels.OutputFileUploadCondition.task_success))]
    )
    )
batch_service_client.task.add_collection(job_id, tasks)

Мониторинг задач

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

Существует множество подходов к мониторингу выполнения задач. Функция wait_for_tasks_to_complete в этом примере использует объект TaskState для отслеживания задач определенного состояния в данном случае завершенного состояния в течение времени.

while datetime.datetime.now() < timeout_expiration:
    print('.', end='')
    sys.stdout.flush()
    tasks = batch_service_client.task.list(job_id)

    incomplete_tasks = [task for task in tasks if
                        task.state != batchmodels.TaskState.completed]
    if not incomplete_tasks:
        print()
        return True
    else:
        time.sleep(1)
...

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

После выполнения задач приложение автоматически удаляет контейнер входного хранилища, который оно создало, а также предоставляет возможность удалить пул и задания Batch. Классы JobOperations и PoolOperations BatchClient имеют методы удаления, которые вызываются при подтверждении удаления. Вы не оплачиваете задания и задачи, но платите за используемые вычислительные узлы. Поэтому рекомендуется выделять пулы только при необходимости. При удалении пула удаляются все выходные данные задачи в узлах. Однако входные и выходные файлы остаются в учетной записи хранения.

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

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

Из этого руководства вы узнали, как:

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