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

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

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

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

Предпосылки

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

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

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

Предоставьте доступ к вашим пакетным и хранилищим аккаунтам

В этом руководстве показано, как аутентифицироваться в пакетная служба Azure и служба хранилища Azure, используя Microsoft Entra ID с DefaultAzureCredential. Приложение не использует ключи аккаунта. Перед запуском приложения убедитесь, что используемая вами личность содержит необходимые роли на обоих аккаунтах.

1. Войдите с помощью Azure CLI. DefaultAzureCredential автоматически обнаруживает этот вход в систему:

   az login
   

2. Назначьте учетной записи пользователя роль, которая позволяет выполнять операции на уровне плоскости данных в учетной записи Batch, например пакетная служба Azure Data Contributor. Эта роль необходима для создания пулов, рабочих мест и задач. Вы можете назначить роль на странице Управление доступом (IAM) учетной записи Batch на портале Azure или с помощью Azure CLI:

   az role assignment create \
       --assignee "<your-user-principal-name>" \
       --role "Azure Batch Data Contributor" \
       --scope "/subscriptions/<subscription-id>/resourceGroups/<resource-group>/providers/Microsoft.Batch/batchAccounts/<batch-account-name>"
   

3. Назначьте своей учетной записи пользователя роль Storage Blob Data Contributor для учетной записи хранения. Эта роль необходима для создания контейнеров, загрузки входных файлов и запроса ключа для делегирования пользователя, который подписывает URL-адреса сигнатуры общего доступа (SAS), используемые задачами:

   az role assignment create \
       --assignee "<your-user-principal-name>" \
       --role "Storage Blob Data Contributor" \
       --scope "/subscriptions/<subscription-id>/resourceGroups/<resource-group>/providers/Microsoft.Storage/storageAccounts/<storage-account-name>"
   

4. Обратите внимание на следующие значения, которые вы добавляете в config.py файл образца в следующем разделе. Вы можете найти их на странице обзора каждого аккаунта в портале Azure:

   • Имя учетной записи пакетной обработки
   • URL-адрес учётной записи пакетной обработки, например https://mybatchaccount.westus2.batch.azure.com
   • Название учетной записи хранилища

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

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

Скачайте или клонируйте пример приложения с 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 и хранилища, указав уникальные имена ваших учётных записей. В примере используется DefaultAzureCredential для проверки подлинности, поэтому ключи учетных записей больше не требуются. Рассмотрим пример.

_BATCH_ACCOUNT_NAME = 'yourbatchaccount'
_BATCH_ACCOUNT_URL = 'https://yourbatchaccount.yourbatchregion.batch.azure.com'
_STORAGE_ACCOUNT_NAME = 'mystorageaccount'

Убедитесь, что вы выполнили вход с помощью az login и что вашей учетной записи назначены роли, описанные в разделе Предоставление доступа к учетным записям Batch и Storage. DefaultAzureCredentialтакже можно найти другие источники учетных данных, такие как управляемая идентичность, Visual Studio Code или переменные среды.

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

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

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 reached the 'Completed' state 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 клиентов

В примере выполняется аутентификация для служб Storage и Batch с помощью DefaultAzureCredential из пакета azure-identity. DefaultAzureCredential последовательно пробует несколько типов учетных данных (переменные среды, управляемое удостоверение, вход через Azure CLI и т. д.), благодаря чему один и тот же код работает в локальной среде разработки и в рабочей среде без необходимости хранить ключи учетных записей.

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

Выборка импортирует следующие типы идентичности и хранилища, а также считывает имена аккаунтов из config.py:

import config
from azure.identity import DefaultAzureCredential
from azure.storage.blob import (
    BlobServiceClient,
    BlobSasPermissions,
    ContainerSasPermissions,
    generate_blob_sas,
    generate_container_sas,
)

credential = DefaultAzureCredential()

blob_service_client = BlobServiceClient(
    account_url=f"https://{config._STORAGE_ACCOUNT_NAME}.blob.core.windows.net/",
    credential=credential)

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

batch_client = BatchClient(
    endpoint=config._BATCH_ACCOUNT_URL,
    credential=credential)

Пакетные вычислительные узлы получают доступ к входным и выходным контейнерам с помощью URL с общей подписью доступа (SAS). Поскольку приложение не использует ключ учётной записи хранения, оно не может подписывать SAS-токены с помощью этого ключа. Вместо этого приложение запрашивает ключ для делегирования пользователя у сервиса Blob, который подписан учетными данными Microsoft Entra приложения, и использует этот ключ для генерации токенов SAS. Дополнительные сведения см. в разделе Создание SAS для делегирования пользователя.

start = datetime.datetime.now(datetime.timezone.utc)
expiry = start + datetime.timedelta(hours=4)
user_delegation_key = blob_service_client.get_user_delegation_key(
    key_start_time=start, key_expiry_time=expiry)

# Sign the SAS tokens with the same expiry as the user delegation key.
sas_expiry = expiry

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

После создания входных и выходных контейнеров blob_service_clientприложение загружает каждый локальный MP4-файл из папки InputFiles в контейнер входа. Следующий upload_file_to_container помощник загружает один файл, генерирует для него токен SAS только для чтения, подписанный ключом делегирования пользователя, и возвращает объект Batch ResourceFile , URL которого содержит токен SAS, чтобы Batch позже мог загрузить файл на вычислительный узел. Приложение вызывает этот помощник один раз для каждого вводного файла:

def upload_file_to_container(blob_service_client, user_delegation_key,
                             sas_expiry, container_name, file_path):
    blob_name = os.path.basename(file_path)
    blob_client = blob_service_client.get_blob_client(container_name, blob_name)

    with open(file_path, "rb") as data:
        blob_client.upload_blob(data, overwrite=True)

    sas_token = generate_blob_sas(
        account_name=config._STORAGE_ACCOUNT_NAME,
        container_name=container_name,
        blob_name=blob_name,
        user_delegation_key=user_delegation_key,
        permission=BlobSasPermissions(read=True),
        expiry=sas_expiry)

    sas_url = f"{blob_client.url}?{sas_token}"

    return models.ResourceFile(http_url=sas_url, file_path=blob_name)

Приложение также генерирует SAS-URL для выходного контейнера, который предоставляет доступ к записи. Задачи используют этот URL для загрузки своих выходных файлов в хранилище:

sas_token = generate_container_sas(
    account_name=config._STORAGE_ACCOUNT_NAME,
    container_name=output_container_name,
    user_delegation_key=user_delegation_key,
    permission=ContainerSasPermissions(write=True, create=True, list=True),
    expiry=sas_expiry)

output_container_sas_url = (
    f"https://{config._STORAGE_ACCOUNT_NAME}.blob.core.windows.net/"
    f"{output_container_name}?{sas_token}")

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

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

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

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

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

new_pool = models.BatchPoolCreateOptions(
    id=pool_id,
    virtual_machine_configuration=models.VirtualMachineConfiguration(
        image_reference=models.BatchVmImageReference(
            publisher="canonical",
            offer="0001-com-ubuntu-server-jammy",
            sku="22_04-lts",
            version="latest"
        ),
        node_agent_sku_id="batch.node.ubuntu 22.04"),
    vm_size=_POOL_VM_SIZE,
    target_dedicated_nodes=_DEDICATED_POOL_NODE_COUNT,
    target_low_priority_nodes=_LOW_PRIORITY_POOL_NODE_COUNT,
    start_task=models.BatchStartTask(
        command_line="/bin/bash -c \"apt-get update && apt-get install -y ffmpeg\"",
        wait_for_success=True,
        user_identity=models.UserIdentity(
            auto_user=models.AutoUserSpecification(
                scope=models.AutoUserScope.POOL,
                elevation_level=models.ElevationLevel.ADMIN)),
    )
)
batch_client.create_pool(pool=new_pool)

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

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

job = models.BatchJobCreateOptions(
    id=job_id,
    pool_info=models.BatchPoolInfo(pool_id=pool_id))

batch_client.create_job(job=job)

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

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

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

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

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(models.BatchTaskCreateOptions(
        id='Task{}'.format(idx),
        command_line=command,
        resource_files=[input_file],
        output_files=[models.OutputFile(
            file_pattern=output_file_path,
            destination=models.OutputFileDestination(
                container=models.OutputFileBlobContainerDestination(
                    container_url=output_container_sas_url)),
            upload_options=models.OutputFileUploadConfiguration(
                upload_condition=models.OutputFileUploadCondition.TASK_SUCCESS))]
    )
    )
batch_client.create_tasks(job_id=job_id, task_collection=tasks)

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

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

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

while datetime.datetime.now() < timeout_expiration:
    print('.', end='')
    sys.stdout.flush()
    tasks = batch_client.list_tasks(job_id=job_id)

    incomplete_tasks = [task for task in tasks if
                        task.state != models.BatchTaskState.COMPLETED]
    if not incomplete_tasks:
        print()
        return True
    else:
        time.sleep(5)
...

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

После выполнения задач приложение автоматически удаляет контейнер входного хранилища, который оно создало, а также предоставляет возможность удалить пул и задания Batch. Методы begin_delete_job и begin_delete_pool класса BatchClient запускают соответствующую операцию удаления при подтверждении запроса. Хотя с вас не взимается плата за сами задания и задачи, с вас взимается плата за вычислительные узлы. Поэтому пулы распределяются только по мере необходимости. При удалении пула удаляются все выходные данные задачи в узлах. Но выходные файлы сохраняются в учетной записи хранения.

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

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

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

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