Samouczek: uruchamianie równoległego obciążenia przy użyciu usługi Azure Batch przy użyciu interfejsu API języka Python

Usługa Azure Batch umożliwia wydajne uruchamianie zadań wsadowych obliczeń równoległych i obliczeń o wysokiej wydajności (HPC, large-scale parallel and high-performance computing) na platformie Azure. Ten samouczek przedstawia przykład równoległego uruchamiania zadań z użyciem technologii Batch w języku Python. Poznasz powszechny przepływ pracy aplikacji Batch oraz jak współpracować programistycznie z zasobami Batch i Storage.

Uwierzytelnij się za pomocą kont Batch i Storage.
Przekazywanie plików wejściowych do usługi Storage.
Utwórz pulę węzłów obliczeniowych, aby uruchomić aplikację.
Utwórz zadanie oraz zadania do przetwarzania plików wejściowych.
Monitorowanie wykonywania zadań.
Pobieranie plików wyjściowych.

W tym samouczku przekonwertujesz pliki multimedialne MP4 na format MP3, równolegle przy użyciu narzędzia open source ffmpeg .

Jeśli nie masz jeszcze konta platformy Azure, przed rozpoczęciem utwórz bezpłatne konto.

Wymagania wstępne

Python w wersji 3.8 lub nowszej
menedżer pakietów pip
Konto usługi Azure Batch i połączone konto usługi Azure Storage. Aby utworzyć te konta, zobacz przewodniki szybkiego startu dla portalu Azure lub Azure CLI.

Zaloguj się do portalu Azure.

Uzyskiwanie poświadczeń konta

W tym przykładzie należy podać poświadczenia dla kont usług Batch i Storage. Najprościej jest uzyskać wymagane poświadczenia w witrynie Azure Portal. (Te poświadczenia możesz również uzyskać za pomocą interfejsów API platformy Azure lub narzędzi wiersza polecenia).

Wybierz opcję Wszystkie usługi>Konta usługi Batch, a następnie wybierz nazwę swojego konta Batch.
Aby wyświetlić poświadczenia usługi Batch, wybierz pozycję Klucze. Skopiuj wartości z pól Konto usługi Batch, Adres URL i Podstawowy klucz dostępu do edytora tekstów.
Aby wyświetlić nazwę i klucze konta pamięci masowej, wybierz pozycję Konto pamięci masowej. Skopiuj wartości z pól Nazwa konta usługi Storage i Klucz1 do edytora tekstów.

Pobieranie i uruchamianie przykładowej aplikacji

Pobieranie przykładowej aplikacji

Pobierz lub sklonuj przykładową aplikację z usługi GitHub. Aby sklonować repozytorium przykładowej aplikacji za pomocą klienta Git, użyj następującego polecenia:

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

Przejdź do katalogu zawierającego plik batch_python_tutorial_ffmpeg.py.

W środowisku języka Python zainstaluj wymagane pakiety przy użyciu polecenia pip.

pip install -r requirements.txt

Użyj edytora kodu, aby otworzyć config.py pliku. Zaktualizuj ciągi znaków dotyczące poświadczeń usługi Batch i konta magazynu, używając wartości unikatowych dla swoich kont. Na przykład:

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

Uruchom aplikację

Aby uruchomić skrypt:

python batch_python_tutorial_ffmpeg.py

Po uruchomieniu aplikacji przykładowej dane wyjściowe w konsoli będą wyglądać mniej więcej następująco. W czasie wykonywania nastąpi wstrzymanie operacji w momencie wyświetlenia komunikatu Monitoring all tasks for 'Completed' state, timeout in 00:30:00... podczas uruchamiania węzłów obliczeniowych puli.

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

Przejdź do konta usługi Batch w witrynie Azure Portal, aby monitorować pulę, węzły obliczeniowe, zadanie i zadania podrzędne. Aby na przykład wyświetlić mapę cieplną węzłów obliczeniowych w puli, wybierz pozycję Pule>LinuxFFmpegPool.

Podczas wykonywania zadań mapa cieplna wygląda następująco:

Zrzut ekranu przedstawiający mapę cieplną puli.

Typowy czas wykonywania wynosi około 5 minut po uruchomieniu aplikacji w domyślnej konfiguracji. Tworzenie puli zajmuje najwięcej czasu.

Pobieranie plików wyjściowych

Przy użyciu witryny Azure Portal można pobrać wyjściowe pliki MP3 wygenerowane przez zadania ffmpeg.

Kliknij Wszystkie usługi>Konta magazynu i kliknij nazwę swojego konta magazynu.
Kliknij Blob>wyjście.
Kliknij prawym przyciskiem myszy jeden z wyjściowych plików MP3, a następnie kliknij polecenie Pobierz. Postępuj zgodnie z monitami wyświetlanymi w przeglądarce, aby otworzyć lub zapisać plik.

Pobieranie pliku wyjściowego

Mimo że nie pokazano tego w tym przykładzie, pliki można również pobrać programowo z węzłów obliczeniowych lub z kontenera magazynu.

Przeglądanie kodu

W poniższych sekcjach przykładowa aplikacja jest rozłożona na kroki dotyczące przetwarzania obciążenia w usłudze Batch. Zapoznaj się z kodem języka Python podczas czytania pozostałej części tego artykułu, ponieważ nie omówiono każdego wiersza kodu w przykładzie.

Uwierzytelnianie klientów obiektów Blob i klientów Batch

Aby komunikować się z kontem magazynu, aplikacja używa pakietu azure-storage-blob do utworzenia obiektu BlockBlobService.

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

Aplikacja tworzy obiekt BatchServiceClient w celu tworzenia pul, zadań i zadań podrzędnych w usłudze Batch oraz zarządzania nimi. Klient usługi Batch w przykładzie używa uwierzytelniania klucza współużytkowanego. Usługa Batch obsługuje również uwierzytelnianie za pośrednictwem identyfikatora Entra firmy Microsoft w celu uwierzytelniania poszczególnych użytkowników lub aplikacji nienadzorowanej.

credentials = batchauth.SharedKeyCredentials(_BATCH_ACCOUNT_NAME,
                                             _BATCH_ACCOUNT_KEY)

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

Przekazywanie plików wejściowych

Aplikacja używa blob_client odwołania do utworzenia kontenera przechowywania dla wejściowych plików MP4 i kontenera dla danych wyjściowych zadania. Następnie wywołuje upload_file_to_container funkcję w celu przekazania plików MP4 w lokalnym katalogu InputFiles do kontenera. Pliki w magazynie są definiowane jako obiekty ResourceFile usługi Batch, które następnie mogą być pobierane przez tę usługę do węzłów obliczeniowych.

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]

Tworzenie puli węzłów obliczeniowych

Następnie w przykładzie tworzona jest pula węzłów obliczeniowych na koncie usługi Batch z wywołaniem funkcji create_pool. Ta zdefiniowana funkcja używa klasy Batch PoolAddParameter do ustawiania liczby węzłów, rozmiaru maszyny wirtualnej i konfiguracji puli. W tym miejscu obiekt VirtualMachineConfiguration określa ImageReference do obrazu Ubuntu Server 20.04 LTS opublikowane w witrynie Azure Marketplace. Usługa Batch obsługuje szeroki zakres obrazów maszyn wirtualnych z witryny Azure Marketplace oraz niestandardowe obrazy maszyn wirtualnych.

Liczba węzłów i rozmiar maszyny wirtualnej są ustawiane przy użyciu zdefiniowanych stałych. Usługa Batch obsługuje dedykowane węzły i węzły typu spot, a w pulach można używać obu tych węzłów. Dla Twojej puli są zarezerwowane węzły dedykowane. Węzły typu spot są oferowane w obniżonej cenie z nadwyżkowej pojemności maszyn wirtualnych na platformie Azure. Węzły typu spot stają się niedostępne, jeśli platforma Azure nie ma wystarczającej pojemności. Przykład domyślnie tworzy pulę zawierającą tylko pięć węzłów typu spot o rozmiarze Standard_A1_v2.

Oprócz właściwości węzła fizycznego ta konfiguracja puli zawiera obiekt StartTask . Polecenie StartTask jest wykonywane na każdym węźle, gdy węzeł dołącza do puli, a za każdym razem, gdy węzeł zostanie ponownie uruchomiony. W tym przykładzie polecenie StartTask uruchamia polecenia powłoki Bash w celu zainstalowania pakietu ffmpeg i zależności w węzłach.

Metoda pool.add przesyła pulę do usługi Batch.

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)

Utwórz pracę

Zadanie usługi Batch określa pulę do uruchamiania zadań oraz opcjonalne ustawienia, takie jak priorytet i harmonogram pracy. Przykładowa aplikacja tworzy zadanie z wywołaniem create_job. Ta zdefiniowana funkcja używa klasy JobAddParameter do utworzenia zadania na twojej puli. Metoda job.add przesyła pulę do usługi Batch. Początkowo praca nie ma zadań.

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

batch_service_client.job.add(job)

Tworzenie zadań

Aplikacja tworzy zadania w zadaniu za pomocą wywołania metody add_tasks. Ta zdefiniowana funkcja tworzy listę obiektów zadań przy użyciu klasy TaskAddParameter . Każde zadanie uruchamia narzędzie ffmpeg w celu przetworzenia obiektu wejściowego resource_files przy użyciu parametru command_line . Narzędzie ffmpeg było już zainstalowane na wszystkich węzłach podczas tworzenia puli. Tutaj wiersz polecenia jest używany do uruchomienia narzędzia ffmpeg w celu przekonwertowania każdego z plików wejściowych w formacie MP4 (wideo) na format MP3 (audio).

Przykładowa aplikacja tworzy obiekt OutputFile dla pliku MP3 po uruchomieniu wiersza polecenia. Pliki wyjściowe każdego zadania (w tym przypadku jeden) są przekazywane do kontenera w połączonym koncie magazynowym, używając właściwości zadania output_files.

Następnie aplikacja dodaje zadania do zadania za pomocą metody task.add_collection , która kolejkuje je do uruchomienia w węzłach obliczeniowych.

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)

Monitorowanie zadań

Gdy zadania są dodawane do zadania, usługa Batch automatycznie kolejkuje je i planuje ich wykonywanie w węzłach obliczeniowych w skojarzonej puli. Na podstawie podanych ustawień usługa Batch obsługuje wszystkie zadania kolejkowania, planowania, ponawiania i innych zadań administracyjnych.

Istnieje wiele podejść do monitorowania wykonywania zadań. Funkcja wait_for_tasks_to_complete w tym przykładzie używa obiektu TaskState do monitorowania zadań pod kątem określonego stanu, w tym przypadku stanu ukończonego w ramach limitu czasu.

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

Czyszczenie zasobów

Po wykonaniu zadań aplikacja automatycznie usuwa utworzony wejściowy kontener magazynu. Daje również możliwość usunięcia puli Batch i zadania. Klasy JobOperations i PoolOperations klasy BatchClient mają metody usuwania, które są wywoływane w przypadku potwierdzenia usunięcia. Mimo że nie są naliczane opłaty za same zadania i zadania podrzędne, są naliczane opłaty za węzły obliczeniowe. W związku z tym zaleca się przydzielanie pul stosownie do potrzeb. Gdy usuniesz pulę, wszystkie dane wyjściowe zadań na węzłach zostaną usunięte. Jednak pliki wejściowe i wyjściowe pozostają na koncie magazynu.

Gdy grupa zasobów, konto usługi Batch i konto magazynu nie będą już potrzebne, usuń je. Aby to zrobić w witrynie Azure Portal, wybierz grupę zasobów dla konta usługi Batch i wybierz pozycję Usuń grupę zasobów.

Następne kroki

W tym samouczku nauczyłeś się następujących rzeczy:

Uwierzytelnij się za pomocą kont Batch i Storage.
Przekazywanie plików wejściowych do usługi Storage.
Utwórz pulę węzłów obliczeniowych, aby uruchomić aplikację.
Utwórz zadanie oraz zadania do przetwarzania plików wejściowych.
Monitorowanie wykonywania zadań.
Pobieranie plików wyjściowych.

Aby uzyskać więcej przykładów użycia interfejsu API języka Python do planowania i przetwarzania obciążeń usługi Batch , zobacz Przykłady języka Python usługi Batch w witrynie GitHub.

Sprzężenie zwrotne

Czy ta strona była pomocna?

Last updated on 2025-04-04