Dataflow ワーカー VM を構成する

このドキュメントでは、Dataflow ジョブのワーカー VM を構成する方法について説明します。

デフォルトでは、Dataflow は、ジョブを実行するワーカー VM のマシンタイプ、および Persistent Disk のサイズとタイプを選択します。ワーカー VM を構成するには、ジョブを作成するときに次のパイプライン オプションを設定します。

マシンタイプ

ワーカー VM の起動時に Dataflow によって使用される Compute Engine マシンタイプ。x86 マシンタイプまたは Arm マシンタイプ（カスタム マシンタイプを含む）を使用できます。

Java

workerMachineType パイプライン オプションを設定する。

Python

machine_type パイプライン オプションを設定する。

Go

worker_machine_type パイプライン オプションを設定する。

カスタム マシンタイプ

カスタム マシンタイプを指定するには、FAMILY-vCPU-MEMORY の形式を使用します。次のように置き換えます。

  • FAMILY: 次のいずれかの値を使用できます。
    マシンシリーズ値
    N1custom
    N2n2-custom
    N2Dn2d-custom
    N4

    ストリーミング ジョブの場合、Streaming Engine を有効にする必要があります。
    バッチジョブの場合、Dataflow Shuffle を有効にする必要があります（デフォルト）。    		 n4-custom
    E2e2-custom
  • vCPU: vCPU の数。
  • MEMORY: メモリ（MB）。

拡張メモリを有効にするには、マシンタイプに -ext を追加します。例: n2-custom-6-3072、n2-custom-2-32768-ext

有効なカスタム マシンタイプの詳細については、Compute Engine ドキュメントのカスタム マシンタイプをご覧ください。

ディスクタイプ

使用する Persistent Disk のタイプ。

Streaming Engine または N4 マシンタイプを使用する場合は、Persistent Disk を指定しないでください。

Java

workerDiskType パイプライン オプションを設定する。

Python

worker_disk_type パイプライン オプションを設定する。

Go

disk_type パイプライン オプションを設定する。

ディスクタイプを指定するには、compute.googleapis.com/projects/PROJECT_ID/zones/ZONE/diskTypes/DISK_TYPE の形式を使用します。

次のように置き換えます。

  • PROJECT_ID: プロジェクト ID
  • ZONE: Persistent Disk のゾーン（例: us-central1-b）。
  • DISK_TYPE: ディスクタイプ（pd-ssd または pd-standard）

詳細については、diskTypes に関する Compute Engine API リファレンス ページをご覧ください。

ディスクサイズ

Persistent Disk のサイズ。

Java

diskSizeGb パイプライン オプションを設定する。

Python

disk_size_gb パイプライン オプションを設定する。

Go

disk_size_gb パイプライン オプションを設定する。

このオプションを設定する場合は、ワーカー ブートイメージとローカルログを考慮して、少なくとも 30 GB を指定します。

ディスクサイズを小さくすると、使用可能なシャッフル I/O が減少します。Dataflow Shuffle や Streaming Engine を使用しないシャッフル バインドされているジョブは、ランタイムとジョブコストが増加する可能性があります。

バッチジョブ

Dataflow Shuffle を使用するバッチジョブの場合、このオプションではワーカー VM ブートディスクのサイズを設定します。Dataflow Shuffle を使用しないバッチジョブの場合、このオプションでは、シャッフルされたデータの保存に使用するディスクのサイズを設定します。ブートディスクのサイズには影響しません。

バッチジョブが Dataflow Shuffle を使用する場合、デフォルトのディスクサイズは 25 GB です。それ以外の場合、デフォルトは 250 GB です。

ストリーミング ジョブ

Streaming Engine を使用するストリーミング ジョブの場合、このオプションでブートディスクのサイズが設定されます。Streaming Engine を使用しないストリーミング ジョブの場合、Dataflow サービスによって作成される追加の Persistent Disk のサイズを設定します。ブートディスクには影響しません。

ストリーミング ジョブが Streaming Engine を使用しない場合は、ブートディスクのサイズをテストフラグ streaming_boot_disk_size_gb で設定できます。たとえば、80 GB のブートディスクを作成するには --experiments=streaming_boot_disk_size_gb=80 を指定します。

ストリーミング ジョブが Streaming Engine を使用する場合、デフォルトのディスクサイズは 30 GB です。それ以外の場合、デフォルトは 400 GB です。

最小 CPU プラットフォーム

特定の CPU 機能に依存するパフォーマンス重視のワークロードがある場合は、ワーカー VM の最小 CPU プラットフォームを指定できます。このオプションを使用すると、Dataflow ワーカーは指定された CPU 世代を満たすか、それを超えるプロセッサを使用します。

最小 CPU プラットフォームを指定するには、min_cpu_platform 試験運用版パイプライン オプションを設定します。値は、選択した CPU プラットフォームの正確な名前（AMD Milan や Intel Ice Lake など）にする必要があります。たとえば、--experiments=min_cpu_platform='AMD Milan' を指定すると、最小 CPU プラットフォームが AMD Milan に設定されます。サポートされている最小 CPU プラットフォームのリストについては、CPU プラットフォームの可用性をご覧ください。制限事項については、最小 CPU プラットフォームを指定する際の制限事項をご覧ください。

指定した最小 CPU プラットフォームで Dataflow ワーカー VM が作成されていることを確認するには、次のようにジョブの Cloud Logging エントリを確認します。

  1. Google Cloud コンソールで Cloud Logging コンソールに移動します。

  2. 次のフィルタを使用し、CPU プラットフォームと Dataflow ジョブ ID の例を、ジョブの情報に置き換えます。

    resource.type="gce_instance"
protoPayload.request.minCpuPlatform="AMD Milan"
"dataflow_job_id"

  3. 結果のログを確認し、VM 作成プロセス中に Dataflow によって最小 CPU プラットフォームが正常に指定されたことを確認します。

Cloud Storage FUSE を使用して Cloud Storage バケットを Dataflow VM にマウントする

Cloud Storage FUSE を使用すると、Cloud Storage バケットを Dataflow VM に直接マウントできます。こうすることで、ソフトウェアがファイルをローカルにあるかのように扱えるようになります。この統合により、データを事前にダウンロードする必要がなくなり、ワークロードのデータアクセスが効率化されます。詳細については、Dataflow と Cloud Storage FUSE を使用して ML データを処理するをご覧ください。

次のステップ