Självstudie: Köra en parallell arbetsbelastning med Azure Batch med hjälp av Python-API:et

Använd Azure Batch för att effektivt köra storskaliga parallella och högpresterande databehandlingsjobb (HPC) i Azure. I den här handledningen går vi igenom ett Python-exempel på att köra en parallell uppgift med Batch. Du lär dig ett vanligt arbetsflöde för Batch-program och hur du interagerar programmatiskt med Batch- och Storage-resurser.

I den här självstudien konverterar du MP4-mediefiler till MP3-format parallellt med hjälp av verktyget ffmpeg med öppen källkod.

Om du inte har något Azure-konto skapar du ett kostnadsfritt konto innan du börjar.

Förutsättningar

• Python version 3.8 eller senare

• pip-pakethanterare

• Ett Azure Batch-konto och ett länkat Azure Storage-konto. Information om hur du skapar dessa konton finns i Snabbstartsguider för Batch för Azure-portalen eller Azure CLI.

Ge tillgång till dina batch- och lagringskonton

Denna handledning visar hur man autentiserar sig till Azure Batch och Azure Storage genom att använda Microsoft Entra ID med DefaultAzureCredential. Appen använder inte kontonycklar. Innan du kör appen, se till att identiteten du använder har de nödvändiga rollerna på båda kontona.

1. Logga in genom att använda Azure CLI. DefaultAzureCredential identifierar automatiskt den här inloggningen:

   az login
   

2. Tilldela ditt användarkonto en roll som tillåter dataplansoperationer på Batch-kontot, såsom Azure Batch Data Contributor. Denna roll krävs för att skapa pooler, jobb och uppgifter. Du kan tilldela rollen på Batch-kontots Access Control (IAM)-sida i Azure-portalen, eller använda 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. Tilldela ditt användarkonto rollen Storage Blob Data Contributor på lagringskontot. Denna roll krävs för att skapa containrar, ladda upp inmatningsfilerna och begära användardelegeringsnyckeln som signerar de delade åtkomstsignaturer (SAS) som används av uppgifterna:

   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. Notera följande värden, som du lägger till i provets config.py-fil i nästa avsnitt. Du kan hitta dem på varje kontos Översiktssida i Azure-portalen:

   • Namn på Batch-kontot
   • Batchkonto-URL, till exempel https://mybatchaccount.westus2.batch.azure.com
   • Lagringskontonamn

Ladda ned och kör exempelappen

Ladda ned exempelprogrammet

Ladda ned eller klona exempelappen från GitHub. Om du vill klona lagringsplatsen för exempelappen med en Git-klient använder du följande kommando:

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

Navigera till katalogen som innehåller filen batch_python_tutorial_ffmpeg.py.

Installera de nödvändiga paketen med hjälp av pipi Python-miljön.

pip install -r requirements.txt

Använd en kodredigerare för att öppna filen config.py. Uppdatera batch- och lagringskontovärdena med namnen som är unika för dina konton. Exemplet använder DefaultAzureCredential för att autentisera, så kontonycklar krävs inte längre. Till exempel:

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

Se till att du är inloggad genom att använda az login och att din identitet har de roller som beskrivs i Grant access till dina batch- och lagringskonton. DefaultAzureCredentialkan också upptäcka andra legitimationskällor, såsom en hanterad identitet, Visual Studio Code eller miljövariabler.

Kör appen

Så här kör du skriptet:

python batch_python_tutorial_ffmpeg.py

När du kör exempelappen ser konsolens utdata ut ungefär så här. Under körningen upplever du en paus vid Monitoring all tasks for 'Completed' state, timeout in 00:30:00... medan poolens beräkningsnoder startas.

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

Gå till Batch-kontot på Azure-portalen för att övervaka poolen, beräkningsnoderna, jobbet och aktiviteterna. Om du till exempel vill se en värmekarta över beräkningsnoderna i poolen väljer du Pooler>LinuxFFmpegPool.

När uppgifter körs ser den termiska kartan ut ungefär så här:

Typisk körningstid är cirka 5 minuter när du kör programmet i standardkonfigurationen. Att skapa poolen är det som tar mest tid.

hämta utdatafilerna.

Du kan använda Azure-portalen till att hämta de utdatafiler i MP3-format som genereras av ffmpeg-uppgifter.

1. Klicka på Alla tjänster>Lagringskonton och sedan på namnet på lagringskontot.
2. Klicka på Blobbar>Utdata.
3. Högerklicka på en av MP3-utdatafilerna och klicka sedan på Ladda ned. Följ anvisningarna i webbläsaren för att öppna eller spara filen.

Även om det inte visas i det här exemplet kan du också ladda ned filerna programmatiskt från beräkningsnoderna eller från lagringscontainern.

Granska koden

I följande avsnitt bryter vi ned exempelprogrammet i de steg som utförs när en arbetsbelastning bearbetas i Batch-tjänsten. Läs Python-koden medan du läser resten av den här artikeln eftersom inte alla kodrader i exemplet diskuteras.

Autentisera Blob- och Batch-klienter

Exemplet autentiserar sig mot både Storage och Batch med hjälp av DefaultAzureCredential från paketet azure-identity. DefaultAzureCredential försöker flera typer av autentiseringsuppgifter i ordning (miljövariabler, hanterad identitet, Azure CLI inloggning och så vidare), vilket gör att samma kod fungerar i lokal utveckling och i produktion utan att lagra kontonycklar.

För att interagera med ett lagringskonto använder appen azure-storage-blob-paketet för att skapa ett BlobServiceClient-objekt som använder autentiseringsuppgifterna.

Exemplet importerar följande identitets- och lagringstyper och läser kontonamnen från 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)

Appen skapar ett BatchClient-objekt för att skapa och hantera pooler, jobb och uppgifter i Batch-tjänsten. Batch-klienten använder samma DefaultAzureCredential för att autentisera via Microsoft Entra ID.

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

Batchberäkningsnoder får tillgång till in- och utmatningscontainrarna genom att använda delade åtkomstsignaturer (SAS) URL:er. Eftersom appen inte använder lagringskontonyckeln kan den inte signera SAS-tokens med den. Istället begär appen en användardelegeringsnyckel från Blob-tjänsten, som är signerad med appens Microsoft Entra-uppgifter, och använder den nyckeln för att generera SAS-tokens. Mer information finns i Skapa en SAS för användardelegering.

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

Ladda upp indatafiler

Efter att den skapat in- och utdatabehållarna med blob_service_client, laddar appen upp varje lokal MP4-fil i InputFiles-mappen till inmatningscontainern. Följande upload_file_to_container hjälpare laddar upp en enda fil, genererar en skrivskyddad SAS-token för den som signeras med användardelegeringsnyckeln, och returnerar ett Batch ResourceFile-objekt vars URL inkluderar SAS-token så att Batch senare kan ladda ner filen till en beräkningsnod. Appen anropar denna hjälpare en gång för varje inmatningsfil:

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)

Appen genererar också en SAS-URL för utdatabehållaren som ger skrivåtkomst. Uppgifterna använder den här URL:en för att ladda upp sina utdatafiler till lagringsplatsen:

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

Skapa en pool med beräkningsnoder

Därefter skapar provet en pool av beräkningsnoder i batchkontot genom att anropa create_pool. Den här definierade funktionen använder klassen Batch BatchPoolCreateOptions för att ange antalet noder, VM-storlek och en poolkonfiguration. I denna konfiguration specificerar ett VirtualMachineConfiguration-objekt en BatchVmImageReference till en Ubuntu Server 22.04 LTS-avbild publicerad i Azure Marketplace. Batch har stöd för ett stort antal Linux- och Windows Server-avbildningar på Azure Marketplace samt för anpassade VM-avbildningar.

Antalet noder och VM-storleken anges med definierade konstanter. Batch har stöd för dedikerade noder och spotnoder, och du kan använda antingen eller båda i dina pooler. Dedikerade noder är reserverade för din pool. Spotnoder erbjuds till ett reducerat pris från överskott av VM-kapacitet i Azure. Spotnoder blir otillgängliga om Azure inte har tillräckligt med kapacitet. Exemplet skapar som standard en pool som bara innehåller fem Spot-noder i storlek Standard_A1_v2.

Förutom egenskaper för fysiska noder innehåller den här poolkonfigurationen ett BatchStartTask-objekt . BatchStartTask körs på varje nod när noden ansluter till poolen och varje gång en nod startas om. I det här exemplet kör BatchStartTask Bash Shell-kommandon för att installera ffmpeg-paketet och beroenden på noderna.

Metoden create_pool skickar poolen till Batch-tjänsten.

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)

Skapa ett jobb

Ett Batch-jobb anger en pool för körning av uppgifter, samt valfria inställningar som en prioritet och ett schema för arbetet. Exemplet skapar ett jobb genom att anropa create_job. Den här definierade funktionen använder klassen BatchJobCreateOptions för att skapa ett jobb i poolen. create_job-metoden skickar jobbet till batchtjänsten. Till en början har jobbet inga uppgifter.

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

batch_client.create_job(job=job)

Skapa uppgifter

Appen skapar uppgifter i jobbet med ett anrop till add_tasks. Den här definierade funktionen skapar en lista över aktivitetsobjekt med klassen BatchTaskCreateOptions . Varje uppgift kör ffmpeg för att bearbeta ett indataobjekt resource_files med hjälp av en command_line parameter. ffmpeg installerades tidigare på varje nod när poolen skapades. Här kör kommandoraden ffmpeg för att konvertera varje MP4-indatafil (video) till en MP3-fil (ljud).

I exemplet skapas ett OutputFile-objekt för MP3-filen när du kör kommandoraden. Utdatarfilerna för varje uppgift (en i det här fallet) laddas upp till en behållare i det länkade lagringskontot med hjälp av aktivitetens output_files egenskap.

Sedan lägger appen till aktiviteter i jobbet med metoden create_tasks, som placerar dem i kö för att köras på beräkningsnoderna.

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)

Övervaka uppgifter

När uppgifter läggs till i ett jobb köar Batch automatiskt och schemalägger dem för exekvering på beräkningsnoder i den associerade poolen. Baserat på de inställningar du anger hanterar Batch all köhantering, schemaläggning, nya försök och andra uppgiftsadministrativa uppgifter.

Det finns många metoder för att övervaka utförande av uppgifter. Funktionen wait_for_tasks_to_complete i det här exemplet använder BatchTaskState-objektet för att övervaka uppgifter för ett visst tillstånd, i det här fallet det slutförda tillståndet, inom en tidsgräns.

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

Rensa resurser

Efter att uppgifterna körts tar appen automatiskt bort den lagringscontainer den skapade, och du får möjligheten att ta bort Batch-poolen och jobbet. Klassens begin_delete_job och begin_delete_pool metoder BatchClient startar varsin raderingsoperation när du bekräftar prompten. Även om du inte debiteras för jobb och uppgifter i sig, debiteras du för beräkningsnoder. Alltså, tilldela bara pooler vid behov. När du tar bort poolen raderas alla aktivitetsutdata på noderna. Utdatafilerna finns dock kvar i lagringskontot.

När de inte längre behövs tar du bort resursgruppen, Batch-kontot och lagringskontot. Om du vill göra det i Azure-portalen väljer du resursgruppen för Batch-kontot och väljer Ta bort resursgrupp.

Nästa steg

I den här handledningen lärde du dig att:

Fler exempel på hur du använder Python API för att schemalägga och bearbeta Batch-arbetsbelastningar finns i Batch Python-exempel på GitHub.