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

Use o Lote do Azure para executar trabalhos em lote paralelos e de computação de alto desempenho (HPC) em grande escala de forma eficiente no Azure. Este tutorial descreve um exemplo em Python de execução de uma carga de trabalho paralela usando o Batch. Aprende um fluxo de trabalho comum de aplicação do Batch e como interagir programaticamente com recursos do Batch e de Armazenamento.

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

Se 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 Azure Batch e uma conta de Armazenamento do Azure vinculada. Para criar essas contas, consulte os guias de início rápido em lote para o portal do Azure ou a CLI do Azure.

Conceder acesso às suas contas do Batch e do Storage

Este tutorial mostra como autenticar-se no Azure Batch e no Armazenamento do Azure ao utilizar o Microsoft Entra ID com DefaultAzureCredential. A aplicação não usa chaves de conta. Antes de executar a aplicação, certifique-se de que a identidade que utiliza tem as funções necessárias em ambas as contas.

1. Inicie sessão utilizando a CLI do Azure. DefaultAzureCredential Capta automaticamente este início de sessão:

   az login
   

2. Atribua à sua conta de utilizador um papel que permita operações de plano de dados na conta Batch, como o Azure Batch Data Contributor. Este papel é necessário para criar pools, trabalhos e tarefas. Pode atribuir a função na página Controlo de acesso (IAM) da conta do Batch no portal do Azure ou utilizar a 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 à sua conta de utilizador o papel de Contribuidor de Dados do Blob de Armazenamento na conta de armazenamento. Este papel é obrigado a criar contentores, carregar os ficheiros de entrada e solicitar a chave de delegação do utilizador que assina as URLs de assinatura de acesso partilhada (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 seguintes, que adiciona ao ficheiro config.py da amostra na secção seguinte. Pode encontrá-los na página de Visão Geral de cada conta no portal Azure:

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

Baixe e execute o aplicativo de exemplo

Descarregar a aplicação de exemplo

Transfira ou clonar a aplicação de exemplo a partir do GitHub. Para clonar o repositório de aplicações de exemplo com um cliente Git, utilize 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.

No 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 do Batch e da conta de armazenamento com os nomes exclusivos das suas contas. O exemplo utiliza o DefaultAzureCredential para autenticar, pelo que as chaves de conta já não são necessárias. Por exemplo:

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

Certifica-te de que tens sessão iniciada utilizando az login e de que a tua identidade tem as funções descritas em Conceder acesso às tuas contas do Batch e do Armazenamento. DefaultAzureCredentialTambém pode descobrir outras fontes de credenciais, como uma identidade gerida, Visual Studio Code ou variáveis de ambiente.

Executar o aplicativo

Para executar o script:

python batch_python_tutorial_ffmpeg.py

Quando executar a aplicação de exemplo, o resultado da consola é semelhante ao seguinte. Durante a execução, ocorre 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

Aceda à conta do Batch no portal do Azure para monitorizar o conjunto, os nós de computação, o trabalho e as tarefas. Por exemplo, para ver um mapa de calor dos nós de computação em seu pool, selecione Pools>LinuxFFmpegPool.

Quando as tarefas estiverem em execução, o mapa térmico é semelhante ao seguinte:

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 do pool demora mais tempo.

Obter ficheiros de saída

Pode utilizar o portal do Azure para transferir os ficheiros MP3 de saída gerados pelas tarefas ffmpeg.

1. Clique em Todos os serviços>Contas de armazenamento e, em seguida, clique no nome da sua conta de armazenamento.
2. Clique em Blobs>resultado.
3. Clique com o botão direito do rato em um dos ficheiros MP3 de saída e, em seguida, clique em Transferir. Siga as instruções no seu browser para abrir ou guardar o ficheiro.

Apesar de não estar mostrado neste exemplo, também pode descarregar os ficheiros programaticamente a partir dos nós de computação ou do contentor de armazenamento.

Reveja o código

As secções seguintes dividem a aplicação de exemplo nos passos que executa para processar uma carga de trabalho no serviço Batch. Consulte o código Python enquanto lê o resto deste artigo, já que nem todas as linhas de código no exemplo são discutidas.

Autenticar os clientes Blob e Batch

O exemplo autentica-se no Storage e no Batch utilizando DefaultAzureCredential do pacote azure-identity. DefaultAzureCredential tenta vários tipos de credenciais por ordem (variáveis de ambiente, identidade gerida, CLI do Azure início de sessão, etc.), o que faz com que o mesmo código funcione no desenvolvimento local e na produção sem armazenar chaves de conta.

Para interagir com uma conta de armazenamento, a aplicação utiliza o pacote azure-storage-blob para criar um objeto BlobServiceClient que utiliza a credencial.

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

A aplicação cria um objeto BatchClient para criar e gerir pools, jobs e tarefas no serviço Batch. O cliente Batch usa o mesmo DefaultAzureCredential para autenticar através de Microsoft Entra ID.

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

Os nós de computação em lote acedem aos contentores de entrada e de saída através de URLs de assinatura de acesso partilhado (SAS). Como a aplicação não usa a chave da conta de armazenamento, não consegue assinar tokens SAS com ela. Em vez disso, a aplicação solicita uma chave de delegação de utilizador ao serviço Blob, que é assinada com as credenciais Microsoft Entra da aplicação, e usa essa chave para gerar os tokens SAS. Para obter mais informações, consulte Criar 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 ficheiros de entrada

Depois de criar os contentores de entrada e saída com blob_service_client, a aplicação carrega cada ficheiro MP4 local na pasta InputFiles para o contentor de entrada. O assistente seguinte upload_file_to_container carrega um único ficheiro, gera um token SAS só de leitura para ele que é assinado com a chave de delegação do utilizador, e devolve um objeto Batch ResourceFile cuja URL inclui o token SAS para que o Batch possa posteriormente descarregar o ficheiro para um nó de computação. A aplicação chama este ajudante uma vez para cada ficheiro 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)

A aplicação também gera um URL SAS para o contentor de saída que concede permissões de escrita. As tarefas usam esta URL para carregar os seus ficheiros 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 conjunto de nós de computação

De seguida, a amostra cria um conjunto de nós de computação na conta Batch ao chamar create_pool. Esta função definida utiliza a classe Batch BatchPoolCreateOptions para definir o número de nós, o tamanho da VM e a configuração do pool. Nesta configuração, um objeto VirtualMachineConfiguration especifica uma BatchVmImageReference para uma imagem Ubuntu Server 22.04 LTS publicada no Azure Marketplace. O Batch suporta inúmeras imagens da VM no Azure Marketplace, bem como imagens da VM personalizadas.

O número de nós e o tamanho da VM são definidos através de constantes definidas. O Batch suporta nós dedicados e nós Spot, e pode-se usar um ou ambos nos seus pools. Os nós dedicados estão reservados para o seu grupo. Os nós spot são oferecidos a um preço reduzido a partir da capacidade excedente de máquinas virtuais no Azure. Os nós spot ficam indisponíveis se o Azure não tiver capacidade suficiente. O exemplo por padrão cria um pool contendo apenas cinco nós Spot em tamanho Standard_A1_v2.

Para além das propriedades físicas do nó, esta configuração de pool inclui um objeto BatchStartTask . O BatchStartTask é executado em cada nó à medida que esse nó entra no pool, e cada vez que um nó é reiniciado. Neste exemplo, o BatchStartTask executa comandos Bash shell para instalar o pacote ffmpeg e as dependências nos nós.

O método create_pool submete o pool ao serviço em 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 emprego

Um trabalho do Batch especifica um pool para executar tarefas e definições opcionais, como uma prioridade e um cronograma para o trabalho. O exemplo cria uma tarefa ao chamar create_job. Esta função definida utiliza a classe BatchJobCreateOptions para criar um trabalho no seu pool. O método create_job submete a tarefa ao serviço em 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. Esta 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 já estava instalado em cada um dos nós quando o pool foi criado. Aqui, a linha de comandos executa o ffmpeg para converter cada ficheiro MP4 (vídeo) de entrada num ficheiro MP3 (áudio).

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

Depois, a aplicação adiciona tarefas ao trabalho com o método create_tasks , que as coloca na fila 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)

Monitorizar tarefas

Quando as tarefas são adicionadas a um trabalho, o Batch enfileira automaticamente e as agenda para execução em nós de computação no pool associado. Com base nas configurações especificadas, o Batch lida com todas as tarefas de enfileiramento de tarefas, agendamento, novas tentativas e outras tarefas de administração de tarefas.

Há muitas abordagens para monitorar a execução de tarefas. A wait_for_tasks_to_complete função neste exemplo usa o objeto BatchTaskState para monitorizar tarefas para um determinado estado, neste caso o estado completado, 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)
...

Limpeza de recursos

Depois de executar as tarefas, a aplicação elimina automaticamente o contentor de armazenamento de entrada que criou e dá-lhe a opção de eliminar o conjunto de recursos e a tarefa do Batch. Os métodos begin_delete_job e begin_delete_pool da BatchClient classe iniciam cada um a operação de eliminação correspondente quando confirmas o prompt. Embora não lhe sejam cobrados os trabalhos e as tarefas em si, ser-lhe-ão cobrados os nós de computação. Assim, aloque os pools apenas conforme necessário. Quando eliminar o grupo, todos os resultados da tarefa nos nós são eliminados. No entanto, os ficheiros de saída permanecem na conta de armazenamento.

Quando já não forem necessários, elimine o grupo de recursos, a conta do Batch e a conta de armazenamento. Para fazer isso no portal do Azure, selecione o grupo de recursos para a conta de lote e escolha Excluir grupo de recursos.

Próximos passos

Neste tutorial, você aprendeu como:

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