Tutorial: Menjalankan beban kerja paralel dengan Azure Batch menggunakan Python API

Gunakan Azure Batch untuk menjalankan pekerjaan batch paralel dan komputasi performa tinggi (HPC) skala besar secara efisien di Azure. Tutorial ini menelusuri contoh Python menjalankan beban kerja paralel menggunakan Batch. Anda mempelajari alur kerja aplikasi Batch umum dan cara berinteraksi secara terprogram dengan sumber daya Batch dan Storage.

Dalam tutorial ini, Anda mengonversi file media MP4 ke format MP3, secara paralel, dengan menggunakan alat sumber terbuka ffmpeg .

Jika Anda tidak memiliki akun Azure, buat akun gratis sebelum memulai.

Prasyarat

• Python versi 3.8 atau yang lebih baru

• manajer paket pip

• Akun Azure Batch dan akun Azure Storage tertaut. Untuk membuat akun ini, lihat panduan mulai cepat Batch untuk portal Microsoft Azure atau Azure CLI.

Berikan akses ke akun Batch dan Storage Anda

Tutorial ini menunjukkan cara mengautentikasi ke Azure Batch dan Azure Storage dengan menggunakan Microsoft Entra ID dengan DefaultAzureCredential. Aplikasi ini tidak menggunakan kunci akun. Sebelum menjalankan aplikasi, pastikan identitas yang Anda gunakan memiliki peran yang diperlukan di kedua akun.

1. Masuk dengan menggunakan Azure CLI. DefaultAzureCredential secara otomatis mendeteksi proses masuk ini:

   az login
   

2. Tetapkan peran pada akun pengguna Anda yang memungkinkan operasi bidang data pada akun Batch, seperti Azure Batch Data Contributor. Peran ini diperlukan untuk membuat kumpulan, pekerjaan, dan tugas. Anda dapat menetapkan peran di halaman Kontrol Akses (IAM) akun Batch di portal Azure, atau menggunakan 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. Tetapkan akun pengguna Anda sebagai Kontributor Data Blob Penyimpanan di akun penyimpanan. Peran ini diperlukan untuk membuat kontainer, mengunggah file input, dan meminta kunci delegasi pengguna yang menandatangani URL tanda tangan akses bersama (SAS) yang digunakan oleh tugas:

   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. Catat nilai berikut, yang Anda tambahkan ke file config.py sampel di bagian berikutnya. Anda dapat menemukannya di halaman Gambaran Umum setiap akun di portal Azure:

   • Nama akun batch
   • URL akun batch, misalnya https://mybatchaccount.westus2.batch.azure.com
   • Nama akun penyimpanan

Unduh dan jalankan aplikasi sampel

Mengunduh aplikasi sampel

Unduh atau klon aplikasi contoh dari GitHub. Untuk mengkloning repo aplikasi sampel dengan klien Git, gunakan perintah berikut:

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

Navigasikan ke direktori yang berisi file batch_python_tutorial_ffmpeg.py.

Di lingkungan Python Anda, instal paket yang diperlukan menggunakan pip.

pip install -r requirements.txt

Gunakan editor kode untuk membuka file config.py. Perbarui nilai akun Batch dan akun penyimpanan dengan nama yang unik bagi masing-masing akun Anda. Sampel menggunakan DefaultAzureCredential untuk mengautentikasi, sehingga kunci akun tidak lagi diperlukan. Contohnya:

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

Pastikan Anda masuk dengan menggunakan az login dan identitas Anda memiliki peran yang dijelaskan dalam Berikan akses ke akun Batch dan Storage Anda. DefaultAzureCredentialjuga dapat menemukan sumber kredensial lain, seperti identitas terkelola, Visual Studio Code, atau variabel lingkungan.

Menjalankan aplikasi

Untuk menjalankan skrip:

python batch_python_tutorial_ffmpeg.py

Saat Anda menjalankan aplikasi sampel, output konsol serupa dengan yang berikut ini. Selama eksekusi, Anda mengalami jeda pada Monitoring all tasks for 'Completed' state, timeout in 00:30:00... ketika node komputasi kumpulan dimulai.

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

Buka akun Batch Anda di portal Microsoft Azure untuk memantau kumpulan, node komputasi, pekerjaan, dan tugas. Misalnya, untuk melihat peta panas simpul komputasi di kumpulan Anda, pilih Kumpulan>LinuxFmpegPool.

Ketika tugas sedang dijalankan, peta panas serupa dengan yang berikut ini:

Waktu eksekusi umum adalah sekitar 5 menit ketika Anda menjalankan aplikasi dalam konfigurasi defaultnya. Pembuatan pool membutuhkan waktu paling lama.

Mengambil file hasil

Anda dapat menggunakan portal Microsoft Azure untuk mengunduh file MP3 output yang dihasilkan oleh tugas ffmpeg.

1. Klik Semua layanan>Akun Storage, lalu klik nama akun penyimpanan Anda.
2. Klik Blobs>output.
3. Klik kanan salah satu file MP3 output, lalu klik Unduh. Ikuti perintah di browser Anda untuk membuka atau menyimpan file.

Meskipun tidak ditampilkan dalam sampel ini, Anda juga dapat mengunduh file secara terprogram dari node komputasi atau dari kontainer penyimpanan.

Meninjau kode

Bagian berikut ini memecah aplikasi sampel ke dalam langkah-langkah yang dijalankan untuk memproses beban kerja di layanan Batch. Lihat kode Python saat Anda membaca sisa artikel ini, karena tidak setiap baris kode dalam sampel dibahas.

Mengautentikasi klien Blob dan Batch

Contoh ini melakukan autentikasi dengan Storage dan Batch menggunakan DefaultAzureCredential dari paket azure-identity. DefaultAzureCredential mencoba beberapa jenis kredensial secara berurutan (variabel lingkungan, identitas terkelola, Azure CLI masuk, dan sebagainya), yang membuat kode yang sama berfungsi dalam pengembangan lokal dan dalam produksi tanpa menyimpan kunci akun.

Untuk berinteraksi dengan akun penyimpanan, aplikasi menggunakan paket azure-storage-blob untuk membuat objek BlobServiceClient yang menggunakan kredensial.

Sampel mengimpor jenis identitas dan penyimpanan berikut, dan membaca nama akun dari 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)

Aplikasi ini membuat objek BatchClient untuk membuat dan mengelola kumpulan, pekerjaan, dan tugas di layanan Batch. Klien Batch menggunakan DefaultAzureCredential yang sama untuk mengautentikasi melalui Microsoft Entra ID.

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

Node komputasi Batch mengakses kontainer input dan output menggunakan URL shared access signature (SAS). Karena aplikasi tidak menggunakan kunci akun penyimpanan, aplikasi tidak dapat menandatangani token SAS dengannya. Sebagai gantinya, aplikasi meminta kunci delegasi pengguna dari layanan Blob, yang ditandatangani dengan kredensial Microsoft Entra aplikasi, dan menggunakan kunci tersebut untuk menghasilkan token SAS. Untuk informasi selengkapnya, lihat Membuat delegasi pengguna SAS.

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

Mengunggah file input

Setelah membuat kontainer input dan output dengan blob_service_client, aplikasi mengunggah setiap file MP4 lokal di folder InputFiles ke penampung input. Pembantu berikut upload_file_to_container mengunggah satu file, menghasilkan token SAS baca-saja untuknya yang ditandatangani dengan kunci delegasi pengguna, dan mengembalikan objek Batch ResourceFile yang URL-nya menyertakan token SAS sehingga Batch nantinya dapat mengunduh file ke simpul komputasi. Aplikasi memanggil pembantu ini sekali untuk setiap file 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)

Aplikasi ini juga menghasilkan URL SAS untuk kontainer output yang memberikan akses tulis. Tugas menggunakan URL ini untuk mengupload file keluaran mereka ke penyimpanan:

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

Membuat kumpulan simpul komputasi

Selanjutnya, sampel membuat kumpulan simpul komputasi di akun Batch dengan memanggil create_pool. Fungsi yang ditentukan ini menggunakan kelas Batch BatchPoolCreateOptions untuk mengatur jumlah simpul, ukuran VM, dan konfigurasi kumpulan. Dalam konfigurasi ini, objek VirtualMachineConfiguration menentukan BatchVmImageReference ke gambar Ubuntu Server 22.04 LTS yang diterbitkan di Marketplace Azure. Batch mendukung berbagai citra VM di Marketplace Azure, serta citra VM kustom.

Jumlah node dan ukuran VM diatur menggunakan konstanta yang ditentukan. Batch mendukung node khusus dan node Spot, dan Anda dapat menggunakan salah satu atau keduanya di kumpulan Anda. Simpul khusus dicadangkan untuk kumpulan Anda. Node Spot ditawarkan dengan harga lebih murah dari kelebihan kapasitas mesin virtual di Azure. Simpul spot tidak tersedia jika Azure tidak memiliki kapasitas yang cukup. Sampel secara default membuat kumpulan yang hanya berisi lima simpul Spot dalam ukuran Standard_A1_v2.

Selain properti simpul fisik, konfigurasi kumpulan ini mencakup objek BatchStartTask . BatchStartTask dijalankan pada setiap simpul saat simpul tersebut bergabung dengan kumpulan, dan setiap kali simpul dimulai ulang. Dalam contoh ini, BatchStartTask menjalankan perintah shell Bash untuk menginstal paket ffmpeg dan dependensi yang diperlukan pada node.

Metode create_pool mengirimkan kumpulan ke layanan 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)

Membuat pekerjaan

Pekerjaan batch menetapkan kumpulan untuk menjalankan tugas dengan pengaturan opsional seperti prioritas dan jadwal untuk pekerjaan tersebut. Contoh tersebut membuat job dengan memanggil create_job. Fungsi yang didefinisikan ini menggunakan kelas BatchJobCreateOptions untuk membuat job pada pool Anda. Metode create_job mengirimkan pekerjaan ke layanan Batch. Awalnya pekerjaan tidak memiliki tugas.

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

batch_client.create_job(job=job)

Membuat tugas

Aplikasi ini membuat tugas dalam pekerjaan dengan panggilan ke add_tasks. Fungsi yang ditentukan ini membuat daftar objek tugas menggunakan kelas BatchTaskCreateOptions . Setiap tugas menjalankan ffmpeg untuk memproses objek input resource_files menggunakan command_line parameter. ffmpeg telah diinstal sebelumnya pada setiap node ketika kumpulan dibuat. Di sini, baris perintah menjalankan ffmpeg untuk mengonversi setiap file MP4 (video) input ke file MP3 (audio).

Sampel membuat objek OutputFile untuk file MP3 setelah menjalankan baris perintah. Setiap file output tugas (satu, dalam hal ini) diunggah ke kontainer di akun penyimpanan tertaut, menggunakan properti tugas output_files .

Kemudian, aplikasi menambahkan tugas ke pekerjaan dengan metode create_tasks , yang mengantrekannya untuk dijalankan pada simpul komputasi.

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)

Memantau tugas

Ketika tugas ditambahkan ke pekerjaan, Batch secara otomatis mengantre dan menjadwalkannya untuk dieksekusi pada simpul komputasi di kumpulan terkait. Berdasarkan pengaturan yang Anda tentukan, Batch menangani semua tugas antrean, penjadwalan, coba lagi, dan tugas administrasi tugas lainnya.

Ada banyak pendekatan untuk memantau eksekusi tugas. Fungsi wait_for_tasks_to_complete dalam contoh ini menggunakan objek BatchTaskState untuk memantau tugas untuk status tertentu, dalam hal ini status selesai, dalam batas waktu.

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

Membersihkan sumber daya

Setelah menjalankan tugas, aplikasi secara otomatis menghapus kontainer penyimpanan input yang dibuatnya, dan memberi Anda opsi untuk menghapus kumpulan dan pekerjaan Batch. Metode begin_delete_job dan BatchClient kelas masing-masing memulai operasi penghapusan yang sesuai saat Anda mengonfirmasi perintah. Meskipun Anda tidak dikenakan biaya untuk pekerjaan dan tugas itu sendiri, Anda dikenakan biaya untuk simpul komputasi. Dengan demikian, alokasikan pool hanya bila diperlukan. Saat Anda menghapus kumpulan, semua output tugas di simpul akan dihapus. Akan tetapi, file output tetap berada di akun penyimpanan.

Saat tidak lagi diperlukan, hapus grup sumber daya, akun Batch, dan akun penyimpanan. Untuk melakukannya di portal Microsoft Azure, pilih grup sumber daya untuk akun Batch dan pilih Hapus grup sumber daya.

Langkah berikutnya

Dalam tutorial ini, Anda belajar cara:

Untuk contoh selengkapnya tentang menggunakan Python API untuk menjadwalkan dan memproses beban kerja Batch, lihat sampel Batch Python di GitHub.