Tutorial: Executar uma carga de trabalho paralela com o Lote do Azure usando a API do 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 percorre um exemplo de Python para executar uma carga de trabalho paralela usando o Lote. 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.

Caso você não tenha uma assinatura do Azure, crie uma conta gratuita do Azure antes de começar.

Pré-requisitos

• Python versão 3.8 ou posterior

• gerenciador de pacotes de 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 no Lote para o portal do Azure ou CLI do Azure.

Entrar no Azure

Entre no portal do Azure.

Obter credenciais da conta

Para este exemplo, você precisa fornecer credenciais para suas contas do Lote e do Armazenamento. Uma maneira simples de obter as credenciais necessárias e no Portal do Azure. (Você também pode obter essas credenciais usando as APIs do Azure ou as ferramentas de linha de comando.)

1. Selecione Todos os serviços>Contas do Lote e, em seguida, selecione o nome de sua conta do Lote.

2. Para ver as credenciais do Lote, selecione Chaves. Copie os valores da conta do Lote, URL, e Chave de acesso primária em um editor de texto.

3. Para ver o nome e as chaves da conta de armazenamento, selecione Conta de armazenamento. Copie os valores de Nome da conta de Armazenamento e Chave1 em um editor de texto.

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 do 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 as cadeias de credenciais da conta de armazenamento e do Lote com os valores exclusivos para suas contas. Por exemplo:

_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=='

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

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 em seu pool, selecione Pools>LinuxFFmpegPool.

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

O tempo de execução típico é de aproximadamente cinco minutos ao executar 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 ao ler o restante deste artigo, pois nem toda linha de código no exemplo é discutida.

Autenticar clientes de Blob e do Lote

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

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

O aplicativo cria um objeto BatchServiceClient para criar e gerenciar pools, trabalhos e tarefas no serviço de Lote. O cliente do Lote no exemplo usa a autenticação de chave compartilhada. O Lote também dá suporte à autenticação por meio do Microsoft Entra ID, para autenticar usuários individuais ou um aplicativo autônomo.

credentials = batchauth.SharedKeyCredentials(_BATCH_ACCOUNT_NAME,
                                             _BATCH_ACCOUNT_KEY)

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

Carregar arquivos de entrada

O aplicativo usa a referência blob_client para criar um contêiner de armazenamento de arquivos MP4 de entrada e um contêiner para a saída da tarefa. Em seguida, ele chama a função upload_file_to_container para carregar arquivos MP4 no diretório InputFiles local para o contêiner. Os arquivos no armazenamento são definidos como objetos ResourceFile do Lote que ele pode baixar mais tarde para os nós de computação.

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]

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

Em seguida, o exemplo cria um pool de nós de computação na conta do Lote com uma chamada para create_pool. Essa função definida usa a classe PoolAddParameter do Lote para definir o número de nós, o tamanho da VM e uma configuração de pool. Aqui, um objeto VirtualMachineConfiguration especifica uma ImageReference para uma imagem do Ubuntu Server 20.04 LTS 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. Nós spot ficam indisponíveis quando o Azure não tem capacidade suficiente. O exemplo, por padrão, cria um pool que contém apenas cinco nós Spot em tamanho Standard_A1_v2.

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

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

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)

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 um trabalho com uma chamada para create_job. Essa função definida usa a classe JobAddParameter para criar um trabalho em seu pool. O método job.add envia o pool ao serviço do Lote. Inicialmente, o trabalho não tem nenhuma tarefa.

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

batch_service_client.job.add(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 TaskAddParameter. Cada tarefa executa ffmpeg para processar um objeto resource_files de entrada usando um parâmetro command_line. 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, neste caso) são carregados em um contêiner na conta de armazenamento vinculada, usando a propriedade output_files da tarefa.

Depois, o aplicativo adiciona tarefas ao trabalho com o método task.add_collection, que as enfileira para execução em 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(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)

Monitorar tarefas

Quando as tarefas são adicionadas a um trabalho, o Lote as enfileira e agenda para execução 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 o monitoramento da execução da tarefa. A função wait_for_tasks_to_complete neste exemplo usa o objeto TaskState para monitorar as tarefas em determinado estado, neste caso, o estado concluído, dentro de um limite de tempo.

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

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. As classes JobOperations e PoolOperations do BatchServiceClient têm métodos de exclusão, chamados se você confirmar a exclusão. Embora você não seja cobrado pelos trabalhos e pelas tarefas, será cobrado pelos nós de computação. Portanto, recomendamos que você aloque os pools conforme a necessidade. Quando você excluir o pool, todas as saídas de tarefa nos nós são excluídas. No entanto, os arquivos de entrada e 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 para a 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 Python no Lote no GitHub.