Tutorial: Executar uma carga de trabalho paralela com o Lote do Azure usando a API Python

Use o Lote do Azure para executar trabalhos em lote de HPC (computação de alto desempenho) e paralelos em larga escala com eficiência no Azure. Este tutorial apresenta um exemplo em Python de execução de uma carga de trabalho paralela usando o Batch. Você conhecerá um fluxo de trabalho de aplicativo comum no Lote e como interagir programaticamente com recursos do Armazenamento e do Lote.

Neste tutorial, você converterá arquivos de mídia MP4 para o formato MP3, em paralelo, usando a ferramenta de código aberto ffmpeg.

Se você ainda não tiver uma conta do Azure, crie uma conta gratuita antes de começar.

Pré-requisitos

• Python versão 3.8 ou posterior

• gerenciador de pacotes pip

• Uma conta do Lote do Azure e uma conta de Armazenamento do Azure vinculada. Para criar essas contas, consulte os guias de início rápido do Batch para o portal do Azure ou a CLI do Azure.

Conceda acesso às suas contas de lote e armazenamento

Este tutorial mostra como se autenticar no Lote do Azure e no Armazenamento do Azure usando o Microsoft Entra ID com DefaultAzureCredential. O app não usa chaves de conta. Antes de executar o aplicativo, certifique-se de que a identidade que você usa possui as funções necessárias em ambas as contas.

1. Faça login usando a CLI do Azure. DefaultAzureCredential Automaticamente capta este login:

   az login
   

2. Atribua à sua conta de usuário uma função que permita operações de data-plane na conta Batch, como Lote do Azure Data Contributor. Essa função é necessária para criar conjuntos, trabalhos e tarefas. Você pode atribuir a função na página de controle de acesso (IAM) da conta Batch no portal do Azure, ou usar a linha de comando do CLI do Azure:

   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. Atribua a função Storage Blob Data Contributor à sua conta de usuário na conta de armazenamento. Esse papel é necessário para criar contêineres, enviar os arquivos de entrada e solicitar a chave de delegação do usuário que assina as URLs de assinatura de acesso compartilhada (SAS) usadas pelas tarefas:

   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. Note os valores a seguir, que você adiciona ao arquivo config.py da amostra na próxima seção. Você pode encontrá-los na página de Visão Geral de cada conta no portal Azure:

   • Nome da conta de lote
   • URL de conta em lote, por exemplo https://mybatchaccount.westus2.batch.azure.com
   • Nome da conta de armazenamento

Baixar e executar o aplicativo de amostra

Baixar o aplicativo de exemplo

Baixe ou clone o aplicativo de exemplo do GitHub. Para clonar o repositório do aplicativo de exemplo com um cliente Git, use o seguinte comando:

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

Navegue até o diretório que contém o arquivo batch_python_tutorial_ffmpeg.py.

Em seu ambiente python, instale os pacotes necessários usando pip.

pip install -r requirements.txt

Use um editor de código para abrir o arquivo config.py. Atualize os valores da conta de armazenamento e do Lote com os nomes exclusivos para suas contas. O exemplo usa DefaultAzureCredential para autenticar, portanto, as chaves de conta não são mais necessárias. Por exemplo:

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

Certifique-se de que você está logado usando az login e que sua identidade tenha as funções descritas em Conceder acesso às suas contas de lote e armazenamento. DefaultAzureCredentialTambém pode descobrir outras fontes de credenciais, como uma identidade gerenciada, Visual Studio Code ou variáveis do ambiente.

Executar o aplicativo

Para executar o script:

python batch_python_tutorial_ffmpeg.py

Quando você executa o aplicativo de exemplo, a saída do console fica mais ou menos assim. Durante a execução, você tem uma pausa em Monitoring all tasks for 'Completed' state, timeout in 00:30:00... enquanto os nós de computação do pool são iniciados.

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

Vá para sua conta do Lote no portal do Azure para monitorar o pool, os nós de computação, os trabalhos e as tarefas. Por exemplo, para ver um mapa de calor dos nós de computação no pool, selecione Pools>LinuxFmpegPool.

Quando as tarefas são executadas, o mapa de calor fica mais ou menos assim:

O tempo de execução típico é de aproximadamente 5 minutos quando você executa o aplicativo em sua configuração padrão. A criação de pool leva mais tempo.

Recuperar arquivos de saída

Você pode usar o Portal do Azure para baixar os arquivos MP3 de saída gerados pelas tarefas de ffmpeg.

1. Clique em Todos os serviços>Contas de armazenamento e, depois, clique no nome da sua conta de armazenamento.
2. Clique em Blobs>saída.
3. Clique com o botão direito em um dos arquivos MP3 de saída e, em seguida, clique em Baixar. Siga as instruções em seu navegador para abrir ou salvar o arquivo.

Embora não seja mostrado neste exemplo, você também pode baixar os arquivos de forma programática a partir dos nós de computação ou do contêiner de armazenamento.

Examine o código

As seções a seguir separa o aplicativo de exemplo nas etapas executadas para processar uma carga de trabalho no serviço Lote. Consulte o código Python enquanto você lê o restante deste artigo, já que nem todas as linhas de código no exemplo são discutidas.

Autenticar clientes de Blob e do Lote

O exemplo se autentica com o Storage e o Batch usando DefaultAzureCredential do pacote azure-identity. DefaultAzureCredential tenta vários tipos de credenciais em ordem (variáveis de ambiente, identidade gerenciada, login na CLI do Azure e assim por diante), o que permite que o mesmo código funcione no desenvolvimento local e em produção sem armazenar chaves de conta.

Para interagir com uma conta de armazenamento, o aplicativo usa o pacote azure-storage-blob para criar um objeto BlobServiceClient que usa a credencial.

O exemplo importa os seguintes tipos de identidade e armazenamento, e lê os nomes das contas de 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)

O aplicativo cria um objeto BatchClient para criar e gerenciar pools, trabalhos e tarefas no serviço Batch. O cliente do Batch usa o mesmo DefaultAzureCredential para se autenticar por meio do Microsoft Entra ID.

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

Os nós de computação do Batch acessam os contêineres de entrada e saída usando URLs de assinatura de acesso compartilhado (SAS). Como o app não usa a chave da conta de armazenamento, ele não pode assinar tokens SAS com ela. Em vez disso, o app solicita uma chave de delegação de usuário do serviço Blob, que é assinada com as credenciais Microsoft Entra do app, e usa essa chave para gerar os tokens SAS. Para obter mais informações, veja Cria uma SAS de delegação de usuário.

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

Carregar arquivos de entrada

Após criar os contêineres de entrada e saída com blob_service_client, o aplicativo faz upload de cada arquivo MP4 local na pasta InputFiles para o contêiner de entrada. O auxiliar a seguir upload_file_to_container faz upload de um único arquivo, gera para ele um token SAS de somente leitura assinado com a chave de delegação do usuário e retorna um objeto Batch ResourceFile cuja URL inclui o token SAS, para que o Batch possa mais tarde baixar o arquivo para um nó de computação. O aplicativo chama esse ajudante uma vez para cada arquivo de entrada:

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)

O aplicativo também gera uma URL SAS para o contêiner de saída que concede acesso de escrita. As tarefas usam esta URL para enviar seus arquivos de saída para armazenamento:

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}")

Criar um pool de nós de computação

Em seguida, a amostra cria um conjunto de nós de computação na conta Batch chamando create_pool. Essa função definida usa a classe BatchPoolCreateOptions do Lote para definir o número de nós, o tamanho da VM e uma configuração de pool. Nessa configuração, um objeto VirtualMachineConfiguration especifica uma Referencia BatchVmImageReference para uma imagem LTS do Ubuntu Server 22.04 publicada no Azure Marketplace. O Lote dá suporte a uma ampla gama de imagens de VM no Azure Marketplace, bem como imagens de VM personalizadas.

O número de nós e o tamanho da VM são definidos usando constantes definidas. O Lote dá suporte a nós dedicados e nós spot, e você pode usar um ou ambos em seus pools. Nós dedicados são reservados para o pool. Nós spot são oferecidos a um preço menor do excedente de capacidade da VM no Azure. Os nós spot ficarão indisponíveis se o Azure não tiver capacidade suficiente. O exemplo, por padrão, cria um pool que contém apenas cinco nós Spot de tamanho Standard_A1_v2.

Além das propriedades do nó físico, essa configuração do pool inclui um objeto BatchStartTask. BatchStartTask é executado em cada nó quando esse nó ingressa no pool, e toda vez que um nó é reiniciado. Neste exemplo, o BatchStartTask executa comandos de shell do Bash para instalar o pacote ffmpeg e as dependências nos nós.

O método create_pool envia o pool para o serviço do Lote.

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)

Criar um trabalho

Um trabalho do Lote especifica um pool onde executar tarefas, e configurações opcionais, como uma prioridade e uma agenda para o trabalho. O exemplo cria uma tarefa ao chamar create_job. Essa função definida usa a classe BatchJobCreateOptions para criar um trabalho em seu pool. O método create_job submete a tarefa ao serviço de lote. Inicialmente, o trabalho não tem tarefas.

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

batch_client.create_job(job=job)

Criar tarefas

O aplicativo cria tarefas no trabalho com uma chamada para add_tasks. Essa função definida cria uma lista de objetos de tarefa usando a classe BatchTaskCreateOptions . Cada tarefa executa ffmpeg para processar um objeto de entrada resource_files usando um command_line parâmetro. O ffmpeg anteriormente foi instalado em cada nó quando o pool foi criado. Aqui, a linha de comando executa ffmpeg para converter cada arquivo MP4 (vídeo) de entrada em um arquivo MP3 (áudio).

O exemplo cria um objeto OutputFile para o arquivo MP3 depois de executar a linha de comando. Os arquivos de saída de cada tarefa (um, nesse caso) são carregados em um contêiner na conta de armazenamento vinculada, usando a propriedade da output_files tarefa.

Em seguida, o aplicativo adiciona tarefas ao trabalho com o método create_tasks , que as enfileira para serem executadas nos nós de computação.

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)

Monitorar tarefas

Quando as tarefas são adicionadas a um trabalho, o Batch as enfileira e agenda automaticamente para serem executadas em nós de computação no pool associado. Com base nas configurações especificadas, o Lote manipula o enfileiramento, o agendamento, a repetição de todas as tarefas e outras obrigações de administração de tarefas.

Há muitas abordagens para monitorar a execução da tarefa. A wait_for_tasks_to_complete função neste exemplo usa o objeto BatchTaskState para monitorar tarefas para um determinado estado, nesse caso, o estado concluído, dentro de um limite de tempo.

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

Limpar os recursos

Depois que ele executa as tarefas, o aplicativo exclui automaticamente o contêiner de armazenamento de entrada criado e oferece a opção de excluir o pool do Lote e o trabalho. Os métodos begin_delete_job e begin_delete_pool da BatchClient classe iniciam a operação de exclusão correspondente quando você confirma o prompt. Embora você não seja cobrado pelos trabalhos e pelas tarefas em si, há cobrança pelos nós de computação. Assim, aloque os pools somente quando necessário. Quando você excluir o pool, todas as saídas de tarefa nos nós são excluídas. No entanto, os arquivos de saída permanecerão na conta de armazenamento.

Quando não forem mais necessário, exclua o grupo de recursos, a conta do Lote e a conta de armazenamento. Para fazer isso no portal do Azure, selecione o grupo de recursos da conta do Lote e escolha Excluir grupo de recursos.

Próximas etapas

Neste tutorial, você aprendeu a:

Para obter mais exemplos de como usar a API do Python para agendar e processar cargas de trabalho do Lote, consulte os exemplos do Python do Lote no GitHub.