Esercitazione: Eseguire un carico di lavoro parallelo con Azure Batch usando l'API Python

Usare Azure Batch per eseguire in modo efficiente processi batch paralleli e HPC (High Performance Computing) su larga scala in Azure. Questa esercitazione illustra un esempio python di esecuzione di un carico di lavoro parallelo con Batch. Vengono fornite informazioni su un flusso di lavoro dell'applicazione Batch comune e su come interagire a livello di codice con le risorse di Archiviazione e Batch.

In questa esercitazione file multimediali MP4 vengono convertiti in parallelo in formato MP3 usando lo strumento open source ffmpeg.

Se non si ha un account Azure, creare un account gratuito prima di iniziare.

Prerequisiti

• Python versione 3.8 o successiva

• gestore di pacchetti pip

• Un account Azure Batch e un account di archiviazione di Azure collegato. Per creare questi account, vedere le guide introduttive di Batch per il portale di Azure o interfaccia della riga di comando di Azure.

Concedi l'accesso ai tuoi account Batch e Storage

Questo tutorial mostra come autenticarsi su Azure Batch e Archiviazione di Azure utilizzando Microsoft Entra ID con DefaultAzureCredential. L'app non usa le chiavi dell'account. Prima di avviare l'app, assicurati che l'identità che usi abbia i ruoli richiesti su entrambi gli account.

1. Accedi usando l'interfaccia della riga di comando di Azure. DefaultAzureCredential rileva automaticamente questo accesso:

   az login
   

2. Assegna al tuo account utente un ruolo che consenta operazioni sul piano dati per l'account Batch, come Azure Batch Data Contributor. Questo ruolo è necessario per creare pool, lavori e compiti. Puoi assegnare il ruolo nella pagina di controllo accesso (IAM) dell'account batch nel portale Azure, oppure usare la CLI di interfaccia della riga di comando di 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. Assegna al tuo account utente il ruolo Storage Blob Data Contributor sull'account di archiviazione. Questo ruolo è necessario per creare container, caricare i file di input e richiedere la chiave di delega utente che firma gli URL della firma di accesso condiviso (SAS) utilizzati dai compiti:

   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. Nota i seguenti valori, che aggiungi al file config.py del campione nella sezione successiva. Puoi trovarli nella pagina panoramica di ogni account nel portale Azure:

   • Nome dell'account batch
   • URL dell'account Batch, ad esempio https://mybatchaccount.westus2.batch.azure.com
   • Nome dell'account di archiviazione

Scaricare ed eseguire l'app di esempio

Scaricare l'app di esempio

Scaricare o clonare l'app di esempio da GitHub. Per clonare il repository dell'app di esempio con un client Git, usare il comando seguente:

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

Passare alla directory che contiene il file batch_python_tutorial_ffmpeg.py.

Nell'ambiente Python installare i pacchetti necessari usando pip.

pip install -r requirements.txt

Usare un editor di codice per aprire il file config.py. Aggiornate i valori di Batch e dell'account di archiviazione con i nomi univoci dei vostri account. L'esempio usa DefaultAzureCredential per l'autenticazione, quindi le chiavi dell'account non sono più necessarie. Per esempio:

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

Assicurati di essere connesso tramite l'uso az login e che la tua identità abbia i ruoli descritti in Concedi accesso ai tuoi account Batch e Storage. DefaultAzureCredentialpuò anche scoprire altre fonti di credenziali, come un'identità gestita, Visual Studio Code o variabili di ambiente.

Eseguire l'app

Per eseguire lo script:

python batch_python_tutorial_ffmpeg.py

Quando si esegue l'applicazione di esempio, l'output della console è simile al seguente. Durante l'esecuzione si verifica una pausa in corrispondenza di Monitoring all tasks for 'Completed' state, timeout in 00:30:00... mentre vengono avviati i nodi di calcolo del pool.

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

Passare all'account Batch nel portale di Azure per monitorare il pool, i nodi di calcolo, il processo e le attività. Ad esempio, per visualizzare una mappa termica dei nodi di calcolo nel pool, selezionare Pool>LinuxFFmpegPool.

Quando le attività sono in esecuzione, la mappa termica è simile all'esempio seguente:

Il tempo di esecuzione tipico è di circa 5 minuti quando si esegue l'applicazione nella configurazione predefinita. La creazione del pool richiede più tempo.

Recupera i file di output

È possibile usare il portale di Azure per scaricare i file MP3 di output generati dalle attività ffmpeg.

1. Fare clic su Tutti i servizi>Account di archiviazione e quindi fare clic sul nome dell'account di archiviazione.
2. Fare clic su BLOB>output.
3. Fare clic con il pulsante destro del mouse su uno dei file MP3 di output e quindi scegliere Scarica. Seguire i prompt nel browser per aprire o salvare il file.

Anche se non viene mostrato in questo esempio, è anche possibile scaricare i file a livello di codice dai nodi di calcolo o dal contenitore di archiviazione.

Esaminare il codice

Nelle sezioni seguenti si esamineranno in dettaglio i singoli passaggi eseguiti dall'applicazione di esempio per l'elaborazione di un carico di lavoro nel servizio Batch. Fare riferimento al codice Python mentre si legge il resto di questo articolo, poiché non viene discussa ogni riga di codice nell'esempio.

Autenticare i client BLOB e Batch

L'esempio si autentica sia con Storage che con Batch usando DefaultAzureCredential del pacchetto azure-identity. DefaultAzureCredential prova più tipi di credenziali in ordine (variabili di ambiente, identità gestita, interfaccia della riga di comando di Azure accesso e così via), che rende lo stesso codice operativo nello sviluppo locale e nell'ambiente di produzione senza archiviare le chiavi dell'account.

Per interagire con un account di archiviazione, l'app usa il pacchetto azure-storage-blob per creare un oggetto BlobServiceClient che usa le credenziali.

L'esempio importa i seguenti tipi di identità e archiviazione, e legge i nomi degli account da 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)

L'app crea un oggetto BatchClient per creare e gestire pool, processi e attività nel servizio Batch. Il client Batch usa lo stesso DefaultAzureCredential per l'autenticazione tramite Microsoft Entra ID.

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

I nodi di calcolo batch accedono ai contenitori di input e output utilizzando URL di firma di accesso condiviso (SAS). Poiché l'app non usa la chiave dell'account storage, non può firmare token SAS con essa. Invece, l'app richiede una chiave di delega utente dal servizio Blob, che viene firmata con le credenziali Microsoft Entra dell'app, e utilizza quella chiave per generare i token SAS. Per altre informazioni, vedere Creare una firma di accesso condiviso di delega utente.

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

Caricare i file di input

Dopo aver creato i contenitori di input e output con blob_service_client, l'app carica ogni file MP4 locale nella cartella InputFiles nel contenitore di input. Il seguente upload_file_to_container helper carica un singolo file, genera un token SAS di sola lettura per esso firmato con la chiave di delega utente e restituisce un oggetto Batch ResourceFile il cui URL include il token SAS, così che Batch possa successivamente scaricare il file su un nodo di calcolo. L'app chiama questo aiutante una volta per ogni file di input:

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)

L'app genera anche un URL SAS per il container di output che concede l'accesso alla scrittura. I compiti utilizzano questo URL per caricare i loro file di output nella memoria:

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

Creare un pool di nodi di calcolo

Successivamente, il campione crea un pool di nodi di calcolo nell'account Batch chiamando create_pool. Questa funzione definita usa la classe Batch BatchPoolCreateOptions per impostare il numero di nodi, le dimensioni della macchina virtuale e una configurazione del pool. In questa configurazione, un oggetto VirtualMachineConfiguration specifica un BatchVmImageReference per un'immagine Ubuntu Server 22.04 LTS pubblicata su Azure Marketplace. Batch supporta una vasta gamma di immagini di VM in Azure Marketplace, oltre che immagini di VM personalizzate.

Il numero di nodi e le dimensioni delle VM vengono impostati usando costanti definite. Batch supporta nodi dedicati e nodi spot e nei pool è possibile usare uno di questi tipi o entrambi. I nodi dedicati sono riservati per il pool. I nodi spot vengono offerti a un prezzo ridotto usando la capacità in eccesso delle VM in Azure. I nodi spot diventano non disponibili se Azure non dispone di capacità sufficiente. L'esempio per impostazione predefinita crea un pool contenente solo cinque nodi spot di dimensioni Standard_A1_v2.

Oltre alle proprietà dei nodi fisici, questa configurazione del pool include un oggetto BatchStartTask . BatchStartTask viene eseguito in ogni nodo perché tale nodo viene aggiunto al pool e ogni volta che un nodo viene riavviato. In questo esempio BatchStartTask esegue i comandi della shell Bash per installare il pacchetto ffmpeg e le dipendenze nei nodi.

Il metodo create_pool invia il pool al servizio Batch.

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)

Crea un lavoro

Un processo di elaborazione batch specifica un pool su cui eseguire le attività e impostazioni facoltative, come una priorità e una pianificazione per il lavoro. L'esempio crea un lavoro chiamando create_job. Questa funzione definita usa la classe BatchJobCreateOptions per creare un processo batch nel tuo pool. Il metodo create_job invia il lavoro al servizio batch. Inizialmente il processo è privo di attività.

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

batch_client.create_job(job=job)

Creare attività

L'app crea le attività nel processo con una chiamata a add_tasks. Questa funzione definita crea un elenco di oggetti attività usando la classe BatchTaskCreateOptions . Ogni attività esegue ffmpeg per elaborare un oggetto di input resource_files usando il parametro command_line. Lo strumento ffmpeg è stato installato in precedenza in ogni nodo al momento della creazione del pool. In questo caso, la riga di comando esegue ffmpeg per convertire ogni file (video) MP4 di input in un file (audio) MP3.

L'esempio crea un oggetto OutputFile per il file MP3 dopo l'esecuzione della riga di comando. I file di output di ogni attività (uno, in questo caso) vengono caricati in un contenitore nell'account di archiviazione collegato, usando la proprietà dell'attività output_files .

L'app aggiunge quindi attività al processo con il metodo create_tasks , che li accoda per l'esecuzione nei nodi di calcolo.

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)

Monitorare le attività

Quando le attività vengono aggiunte a un processo, Batch li accoda automaticamente e li pianifica per l'esecuzione nei nodi di calcolo nel pool associato. In base alle impostazioni specificate, Batch gestisce tutte le operazioni di accodamento, pianificazione, ripetizione di tentativi e tutte le altre operazioni di amministrazione relative alle attività.

Esistono molti approcci per il monitoraggio dell'esecuzione delle attività. La wait_for_tasks_to_complete funzione in questo esempio usa l'oggetto BatchTaskState per monitorare le attività per un determinato stato, in questo caso lo stato completato, entro un limite di 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)
...

Pulire le risorse

Al termine dell'esecuzione delle attività, l'app elimina automaticamente il contenitore di archiviazione di input che ha creato e consente di scegliere se eliminare il pool e il processo di Batch. I metodi begin_delete_job e begin_delete_pool della BatchClient classe avviano ciascuno l'operazione di cancellazione corrispondente quando confermi il prompt. Anche se non ti viene addebitato per i lavori e le attività in sé, ti vengono addebitati i nodi di calcolo. Pertanto, assegna i pool solo quando necessario. Quando si elimina il pool, tutto l'output delle attività nei nodi viene eliminato. I file di output rimangono tuttavia nell'account di archiviazione.

Quando non sono più necessari, eliminare il gruppo di risorse, l'account Batch e l'account di archiviazione. A tale scopo, nel portale di Azure selezionare il gruppo di risorse per l'account Batch e scegliere Elimina gruppo di risorse.

Passaggi successivi

In questo tutorial, hai imparato come:

Per altri esempi sull'uso dell'API Python per pianificare ed elaborare carichi di lavoro batch, vedere gli esempi di Batch Python in GitHub.