チュートリアル: Python API を使用して Azure Batch で並列ワークロードを実行する

Azure Batch を使用すると、大規模な並列コンピューティングやハイパフォーマンスコンピューティング (HPC) のバッチジョブを Azure で効率的に実行することができます。このチュートリアルでは、Batch を使用して並列ワークロードを実行する Python の例について説明します。一般的な Batch アプリケーションワークフローのほか、Batch および Storage のリソースをプログラムで操作する方法を学習します。

Batch アカウントおよびストレージアカウントで認証します。
Storage に入力ファイルをアップロードします。
アプリケーションを実行するコンピューティングノードのプールを作成します。
入力ファイルを処理するジョブとタスクを作成します。
タスクの実行を監視する。
出力ファイルを取得します。

このチュートリアルでは、ffmpeg オープンソースツールを使用して複数の MP4 メディアファイルを並行して MP3 形式に変換します。

Azure アカウントをお持ちでない場合は、開始する前に無料アカウントを作成してください。

[前提条件]

Python バージョン 3.8 以降
pip パッケージマネージャー
Azure Batch アカウントとリンクされた Azure Storage アカウント。これらのアカウントを作成するには、 Azure portal または Azure CLI の Batch クイックスタートガイドを参照してください。

バッチアカウントとストレージアカウントへのアクセスを許可してください

このチュートリアルでは、Microsoft Entra IDとDefaultAzureCredentialを使ってAzure BatchおよびAzure Storageに認証する方法を示しています。アプリはアカウントキーを使いません。アプリを実行する前に、両方のアカウントで必要な役割を持つIDを確認してください。

Azure CLIを使ってサインインしてください。 DefaultAzureCredential このサインインを自動的に検出します:
```
az login
```
ユーザーアカウントに、Azure Batch Data Contributorのようなバッチアカウントでデータプレーン操作を許可する役割を割り当ててください。このロールは、プール、ジョブ、タスクを作成するために必要です。 Azureポータルのバッチアカウントのアクセス制御(IAM)ページで役割を割り当てるか、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>"
```
ユーザーアカウントにストレージアカウントの Storage Blob Data Contributor 役割を割り当ててください。この役割は、コンテナの作成、入力ファイルのアップロード、そしてタスクで使用される共有アクセス署名(SAS)URLに署名するユーザー委任キーの要求に必要です。
```
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>"
```
次のセクションでサンプルの config.py ファイルに追加する以下の値に注目してください。それらはAzureポータルの各アカウントの概要ページで見つけることができます:
- バッチアカウント名
- 例えばバッチアカウントのURLなどです https://mybatchaccount.westus2.batch.azure.com
- ストレージアカウント名

Note

役割の割り当てが伝播するまで数分かかることもあります。ロールを割り当てた直後に認証エラーでアプリが失敗した場合は、数分待ってから再度試してください。

サンプルアプリケーションのダウンロードと実行

Important

batch-python-ffmpeg-tutorialリポジトリのダウンロード可能なサンプルは、このチュートリアルに合わせて更新されています。そのアップデートが公開されるまでは、リポジトリには以前の鍵ベースの認証やUbuntu 20.04のコードが残っている可能性があります。 この記事のコードが真実の出典です。 ダウンロードしたサンプルがここにあるスニペットと一致しない場合は、この記事で示されたコードを従ってください。

サンプルアプリのダウンロード

GitHub からサンプルアプリをダウンロードまたは複製します。 Git クライアントを使用してサンプルアプリリポジトリを複製するには、次のコマンドを使用します。

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

ファイル batch_python_tutorial_ffmpeg.pyを含むディレクトリに移動します。

Python 環境で、 pipを使用して必要なパッケージをインストールします。

pip install -r requirements.txt

コードエディターを使用して、ファイル config.py を開きます。 Batch とストレージアカウントの値を、アカウントに固有の名前で更新します。このサンプルでは、DefaultAzureCredential を使用して認証を行います。そのため、アカウントキーは不要になりました。例えば次が挙げられます。

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

az loginを使ってサインインしていること、そしてBatchアカウントとStorageアカウントへのアクセス権限付与に記載された役割を身元に持っていることを確認してください。 DefaultAzureCredential管理型アイデンティティ、Visual Studio Code、環境変数など、他の認証情報ソースも発見できます。

アプリを実行する

スクリプトを実行するには、次の手順を実行します。

python batch_python_tutorial_ffmpeg.py

サンプルアプリケーションを実行すると、コンソールの出力は次のようになります。実行中、プールのコンピューティングノードを開始する際に、Monitoring all tasks for 'Completed' state, timeout in 00:30:00... で一時停止が発生します。

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

プール、コンピューティングノード、ジョブ、タスクを監視するには、Azure Portal で Batch アカウントに移動します。たとえば、プール内のコンピューティングノードのヒートマップを表示するには、 プール>LinuxFFmpegPool を選択します。

タスクが実行されていると、ヒートマップは次のようになります。

プールヒートマップのスクリーンショット。

通常の実行時間は、既定の構成でアプリケーションを実行すると約 5 分です 。プールの作成に最も時間がかかります。

出力ファイルを取得する

Azure Portal を使用して、ffmpeg タスクによって生成された出力 MP3 ファイルをダウンロードすることができます。

[すべてのサービス]>[ストレージアカウント] の順にクリックし、ストレージアカウントの名前をクリックします。
[BLOB]>[出力] の順にクリックします。
出力 MP3 ファイルの 1 つを右クリックして、[ダウンロード] をクリックします。ブラウザーのメッセージに従って、ファイルを開くか保存します。

出力ファイルをダウンロードする

このサンプルには示されていませんが、コンピューティングノードまたはストレージコンテナーからプログラムでファイルをダウンロードすることもできます。

コードの確認

以降のセクションでは、サンプルアプリケーションを、Batch サービスでワークロードを処理するために実行する複数の手順に分けます。この記事の残りの部分を読んでいる間は Python コードを参照してください。サンプル内のすべてのコード行が説明されているわけではありません。

BLOB クライアントと Batch クライアントの認証

このサンプルでは、Azure ID パッケージの DefaultAzureCredential を使用して、Storage と Batch の両方で認証を行います。 DefaultAzureCredential は、複数の資格情報の種類 (環境変数、マネージド ID、Azure CLI サインインなど) を順番に試行します。これにより、アカウントキーを格納せずに、ローカル開発と運用環境で同じコードが動作します。

ストレージアカウントを操作するために、アプリは azure-storage-blob パッケージを使用して、資格情報を使用する BlobServiceClient オブジェクトを作成します。

サンプルは以下の識別情報とストレージタイプをインポートし、 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)

アプリは、Batch サービスでプール、ジョブ、タスクを作成および管理するための BatchClient オブジェクトを作成します。 Batch クライアントは、同じ DefaultAzureCredential を使用して、Microsoft Entra IDを介して認証します。

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

バッチ計算ノードは、共有アクセス署名(SAS)URLを使って入力および出力コンテナにアクセスします。アプリはストレージアカウントキーを使わないため、SASトークンに署名できません。代わりに、アプリはBlobサービスからユーザーの委任キーを要求し、その鍵はアプリのMicrosoft Entra認証情報で署名され、そのキーを使ってSASトークンを生成します。詳細については、「ユーザー委任 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

Note

このサンプルのユーザー委任キーは4時間有効です。ユーザー委任キーで署名された SAS トークンは、そのキーの有効期間を超えて有効にすることはできません。また、ユーザー委任キーの有効期間は最長 7 日間です。長時間実行のワークロードの場合は、新しいキーを要求し、SASのURLが期限切れになる前に再生成してください。

入力ファイルのアップロード

blob_service_clientで入力と出力コンテナを作成した後、アプリはInputFilesフォルダ内の各ローカルMP4ファイルを入力コンテナにアップロードします。次の upload_file_to_container ヘルパーは単一のファイルをアップロードし、ユーザー委任キーで署名された読み取り専用のSASトークンを生成し、SASトークンを含むURLを持つバッチ ResourceFile オブジェクトを返します。これにより、Batchは後でそのファイルをコンピュートノードにダウンロードできます。アプリは入力ファイルごとにこのヘルパーを1回呼び出します:

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)

また、アプリは書き込みアクセス権を与える出力コンテナ用のSAS URLも生成します。タスクはこのURLを使って出力ファイルをストレージにアップロードします:

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

コンピューティングノードのプールの作成

次に、サンプルは create_poolを呼び出してバッチアカウント内の計算ノードのプールを作成します。この定義済み関数では、Batch BatchPoolCreateOptions クラスを使用して、ノード数、VM サイズ、プール構成を設定します。この構成では、VirtualMachineConfigurationオブジェクトがAzure Marketplaceに公開されたUbuntu Server 22.04 LTSイメージのBatchVmImageReferenceを指定します。 Batch は、Azure Marketplace のさまざまな VM イメージだけでなく、カスタム VM イメージもサポートしています。

ノードの数と VM のサイズは、定義済みの定数を使用して設定されます。 Batch では専用ノードとスポットノードがサポートされているため、ご利用のプールではそのいずれかまたは両方を使用できます。専用ノードは、プール用に予約されています。スポットノードは、Azure の VM の余剰容量から割引価格で提供されます。 Azure に十分な容量がない場合、スポットノードは使用できなくなります。既定では、このサンプルでは、サイズ がStandard_A1_v2のスポットノードが 5 つだけ含まれるプールが作成されます。

物理ノードのプロパティに加えて、このプール構成には BatchStartTask オブジェクトが含まれます。 BatchStartTask は、そのノードがプールに参加し、ノードが再起動されるたびに、各ノードで実行されます。この例では、BatchStartTask によって Bash シェルコマンドが実行され、ffmpeg パッケージと依存関係がノードにインストールされます。

create_pool メソッドは、プールを 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)

Note

マーケットプレイスのVMイメージやバッチノードエージェントにはサポート終了日があります。 Ubuntu Server 20.04のLTSイメージと batch.node.ubuntu 20.04 ノードエージェントは、新しいバッチプールでサポートされていません。バッチアカウントが現在サポートしている画像参照やノードエージェントのSKUを一覧にするには、 list_supported_images メソッドを呼び出してください。

ジョブの作成

Batch ジョブでは、タスクの実行対象となるプールと、作業の優先順位やスケジュールなどのオプションの設定を指定します。サンプルは create_jobを呼び出してジョブを作成します。この定義済み関数は、BatchJobCreateOptions クラスを使用してプールにジョブを作成します。 create_jobメソッドはジョブをバッチサービスに送信します。最初、ジョブにはタスクがありません。

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

batch_client.create_job(job=job)

タスクの作成

アプリは、add_tasksを呼び出してジョブ内でタスクを作成します。この定義された関数は、 BatchTaskCreateOptions クラスを使用してタスクオブジェクトのリストを作成します。各タスクは ffmpeg を実行し、resource_files パラメーターを使用して入力command_lineオブジェクトを処理します。 ffmpeg は、以前にプールが作成されたときに各ノードにインストールされています。ここでは、コマンドラインで ffmpeg を実行して、各入力 MP4 (ビデオ) ファイルを MP3 (オーディオ) ファイルに変換します。

このサンプルでは、コマンドラインの実行後に MP3 ファイルの OutputFile オブジェクトを作成します。各タスクの出力ファイル (この場合は 1 つ) は、タスクの output_files プロパティを使用して、リンクされたストレージアカウント内のコンテナーにアップロードされます。

次に、アプリは、create_tasks メソッドを使用してジョブにタスクを追加します。これにより、コンピューティングノードで実行するためにキューに入れられます。

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)

タスクの監視

タスクがジョブに追加されると、Batch は自動的にキューに登録し、関連付けられているプール内のコンピューティングノードで実行されるようにスケジュールします。指定した設定に基づいて、Batch はすべてのタスクキュー、スケジュール設定、再試行、およびその他のタスク管理作業を処理します。

タスクの実行を監視するには、多くの方法があります。この例の wait_for_tasks_to_complete 関数は、BatchTaskState オブジェクトを使用して、特定の状態 (この場合は完了した状態) のタスクを制限時間内に監視します。

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

リソースをクリーンアップする

タスクの実行後、自動的に、作成された入力用ストレージコンテナーが削除され、Batch プールとジョブを削除するためのオプションが表示されます。クラスのbegin_delete_jobメソッドとBatchClientメソッドは、プロンプトを確認すると対応する削除操作を開始します。ジョブやタスク自体に料金はかかりませんが、コンピューティングノードに対しては料金がかかります。したがって、プールは必要に応じてのみ割り当ててください。プールを削除すると、ノード上のタスク出力はすべて削除されます。ただし、出力ファイルはストレージアカウントに残ります。

リソースグループ、Batch アカウント、ストレージアカウントは、不要になったら削除します。 Azure portal でこれを行うには、Batch アカウントのリソースグループを選択し、[ リソースグループの削除] を選択します。

次のステップ

このチュートリアルでは、次の方法を学習しました。

Batch アカウントおよびストレージアカウントで認証します。
Storage に入力ファイルをアップロードします。
アプリケーションを実行するコンピューティングノードのプールを作成します。
入力ファイルを処理するジョブとタスクを作成します。
タスクの実行を監視する。
出力ファイルを取得します。

Python API を使用して Batch ワークロードをスケジュールおよび処理するその他の例については、GitHub の Batch Python サンプルを参照してください。

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-07-10

チュートリアル: Python API を使用して Azure Batch で並列ワークロードを実行する

[前提条件]

バッチアカウントとストレージアカウントへのアクセスを許可してください

サンプル アプリケーションのダウンロードと実行

サンプル アプリのダウンロード

アプリを実行する

出力ファイルを取得する

コードの確認

BLOB クライアントと Batch クライアントの認証

入力ファイルのアップロード

コンピューティング ノードのプールの作成

ジョブの作成

タスクの作成

タスクの監視

リソースをクリーンアップする

次のステップ

フィードバック

その他のリソース

サンプルアプリケーションのダウンロードと実行

サンプルアプリのダウンロード

コンピューティングノードのプールの作成