Dataproc サーバーレスで GPU を使用する

GPU アクセラレータを Dataproc サーバーレス バッチ ワークロードに接続すると、次のメリットが得られます。

  • 大規模なデータ分析ワークロードの処理を高速化します。

  • GPU 機械学習ライブラリを使用して、大規模なデータセットでのモデル トレーニングを高速化します。

  • 動画や自然言語処理などの高度なデータ分析を実行します。

すべてのサポートされている Dataproc サーバーレス Spark ランタイムで、各ワークロード ノードに Spark RAPIDS ライブラリが追加されます。Dataproc Serverless Spark ランタイム バージョン 1.1 では、ワークロード ノードに XGBoost ライブラリも追加されます。これらのライブラリは、GPU 高速化ワークロードで使用できる強力なデータ変換ツールと機械学習ツールを提供します。

GPU のメリット

Dataproc サーバーレス Spark ワークロードで GPU を使用するメリットは次のとおりです。

  • パフォーマンスの向上: GPU アクセラレーションにより、特に機械学習やディープラーニング、グラフ処理、複雑な分析など、コンピューティング負荷の高いタスクで Spark ワークロードのパフォーマンスを大幅に向上させることができます。

  • モデル トレーニングの高速化: 機械学習タスクでは、GPU を接続することでモデルのトレーニングに必要な時間を大幅に短縮でき、データ サイエンティストとエンジニアは迅速に反復処理とテストを行うことができます。

  • スケーラビリティ: 処理ニーズが複雑になるにつれて、より多くの GPU ノードまたはより強力な GPU をノードに追加できます。

  • 費用対効果: GPU には初期投資が必要ですが、処理時間の短縮とリソースの効率的な使用により、時間の経過とともに費用を削減できます。

  • 高度なデータ分析: GPU アクセラレーションにより、大規模なデータセットに対して画像や動画の分析、自然言語処理などの高度な分析を実行できます。

  • 製品の改善: 処理速度が向上し、意思決定の迅速化とアプリケーションの応答性の向上が実現します。

制限事項と考慮事項

料金

アクセラレータの料金については、Dataproc サーバーレスの料金をご覧ください。

準備

GPU アクセラレータが接続されたサーバーレス Batch ワークロードを作成する前に、次の操作を行います。

  1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  3. Make sure that billing is enabled for your Google Cloud project.

  4. Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

    Enable the APIs

  5. Install the Google Cloud CLI.

  6. To initialize the gcloud CLI, run the following command: 

    gcloud init

  7. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  8. Make sure that billing is enabled for your Google Cloud project.

  9. Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

    Enable the APIs

  10. Install the Google Cloud CLI.

  11. To initialize the gcloud CLI, run the following command: 

    gcloud init
  12. In the Google Cloud console, go to the Cloud Storage Buckets page.

    Go to Buckets page

  13. Click Create bucket.
  14. On the Create a bucket page, enter your bucket information. To go to the next step, click Continue.
    • For Name your bucket, enter a name that meets the bucket naming requirements.
    • For Choose where to store your data, do the following:
      • Select a Location type option.
      • Select a Location option.
    • For Choose a default storage class for your data, select a storage class.
    • For Choose how to control access to objects, select an Access control option.
    • For Advanced settings (optional), specify an encryption method, a retention policy, or bucket labels.
  15. Click Create.

GPU アクセラレータを使用してサーバーレス Batch ワークロードを作成する

NVIDIA L4 GPU を使用する Dataproc サーバーレス バッチ ワークロードを送信して、並列化された PySpark タスクを実行します。gcloud CLI を使用して次の操作を行います。

  1. [展開する] をクリックし、テキストまたはコードエディタを使用して、一覧表示された PySpark コードを作成し、ローカルマシンの test-py-spark-gpu.py ファイルに保存します。

    #!/usr/bin/env python

"""S8s Accelerators Example."""

import subprocess
from typing import Any
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import IntegerType
from pyspark.sql.types import StructField
from pyspark.sql.types import StructType

spark = SparkSession.builder.appName("joindemo").getOrCreate()


def get_num_gpus(_: Any) -> int:
  """Returns the number of GPUs."""
  p_nvidia_smi = subprocess.Popen(
      ["nvidia-smi", "-L"], stdin=None, stdout=subprocess.PIPE
  )
  p_wc = subprocess.Popen(
      ["wc", "-l"],
      stdin=p_nvidia_smi.stdout,
      stdout=subprocess.PIPE,
      stderr=subprocess.PIPE,
      universal_newlines=True,
  )
  [out, _] = p_wc.communicate()
  return int(out)


num_workers = 5
result = (
    spark.sparkContext.range(0, num_workers, 1, num_workers)
    .map(get_num_gpus)
    .collect()
)
num_gpus = sum(result)
print(f"Total accelerators: {num_gpus}")

# Run the join example
schema = StructType([StructField("value", IntegerType(), True)])
df = (
    spark.sparkContext.parallelize(range(1, 10000001), 6)
    .map(lambda x: (x,))
    .toDF(schema)
)
df2 = (
    spark.sparkContext.parallelize(range(1, 10000001), 6)
    .map(lambda x: (x,))
    .toDF(schema)
)
joined_df = (
    df.select(col("value").alias("a"))
    .join(df2.select(col("value").alias("b")), col("a") == col("b"))
    .explain()
)

  2. ローカルマシンで gcloud CLI を使用して、各ワーカーを L4 GPU で高速化した 5 ワーカーで、Dataproc Serverless サーバーレス バッチジョブを送信します。

    gcloud dataproc batches submit pyspark test-py-spark-gpu.py \
    --project=PROJECT_ID \
    --region=REGION \
    --deps-bucket=BUCKET_NAME \
    --version=1.1 \
    --properties=spark.dataproc.executor.compute.tier=premium,spark.dataproc.executor.disk.tier=premium,spark.dataproc.executor.resource.accelerator.type=l4,spark.executor.instances=5,spark.dataproc.driverEnv.LANG=C.UTF-8,spark.executorEnv.LANG=C.UTF-8,spark.shuffle.manager=com.nvidia.spark.rapids.RapidsShuffleManager

メモ:

  • PROJECT_ID: Google Cloud プロジェクト ID
  • REGION: ワークロードを実行する利用可能な Compute Engine リージョン。
  • BUCKET_NAME: Cloud Storage バケットの名前。Spark は、バッチ ワークロードを実行する前に、ワークロードの依存関係をこのバケットの /dependencies フォルダにアップロードします。
  • --version: すべてのサポートされている Dataproc サーバーレス ランタイムで、GPU 高速化ワークロードの各ノードに RAPIDS ライブラリが追加されます。ランタイム バージョン 1.1 でのみ、GPU 高速化ワークロードの各ノードに XGBoost ライブラリが追加されます。

  • --properties（Spark リソース割り当てプロパティを参照）:

    • spark.dataproc.driverEnv.LANG=C.UTF-8 と spark.executorEnv.LANG=C.UTF-8（2.2 より前のランタイム バージョンで必須）: これらのプロパティは、デフォルトの文字セットを C.UTF-8 に設定します。

    • spark.dataproc.executor.compute.tier=premium（必須）: GPU アクセラレーション ワークロードは、プレミアム Data Compute Unit（DCU）を使用して課金されます。Dataproc サーバーレスのアクセラレータの料金をご覧ください。

    • spark.dataproc.executor.disk.tier=premium（必須）: A100-40、A100-80、または L4 アクセラレータを使用するノードは、プレミアム ディスク ティアを使用する必要があります。

    • spark.dataproc.executor.resource.accelerator.type=l4（必須）: 指定する GPU タイプは 1 つだけです。サンプルジョブでは、L4 GPU が選択されます。次のアクセラレータ タイプでは、次の引数名を指定できます。

      GPU のタイプ 引数名
      A100 40GB a100-40
      A100 80GB a100-80

    • spark.executor.instances=5（必須）: 2 つ以上指定する必要があります。この例では 5 に設定します。

    • spark.executor.cores（省略可）: このプロパティを設定すると、コア vCPU の数を指定できます。L4 GPU の有効な値は、4（デフォルト）、8、12、16、24、48、96 です。A100 GPU で有効なデフォルト値は 12 のみです。L4 GPU と 24、48、または 96 コアを使用する構成では、各エグゼキュータに 2、4、または 8 GPU が接続されます。他のすべての構成には 1 GPU が接続されています。

    • spark.dataproc.executor.disk.size（必須）: L4 GPU のディスクサイズは 375 GB に固定されています。ただし、24、48、96 コアを使用する構成では、それぞれ 750、1,500、3,000 GB になります。L4 高速化ワークロードの送信時に、このプロパティを別の値に設定すると、エラーが発生します。A100 40 または A100 80 GPU を選択した場合、有効なサイズは 375g、750g、1500g、3000g、6000g、9000g です。

    • spark.executor.memory（省略可）と spark.executor.memoryOverhead（省略可）: これらのプロパティのいずれかを設定できますが、両方を設定することはできません。set プロパティで使用されていない使用可能なメモリ量が、unset プロパティに適用されます。デフォルトでは、spark.executor.memoryOverhead は PySpark バッチ ワークロードの使用可能なメモリの 40% に設定され、他のワークロードの場合は 10% に設定されます（Spark のリソース割り当てプロパティをご覧ください）。

      次の表に、A100 と L4 GPU のさまざまな構成で設定できるメモリの最大量を示します。どちらのプロパティの最小値も 1024 MB です。

      A100 (40 GB) A100 (80 GB) L4（4 コア） L4（8 コア） L4（12 コア） L4（16 コア） L4（24 コア） L4（48 コア） L4（96 コア）
      最大合計メモリ（MB） 78040 165080 13384 26768 40152 53536 113072 160608 321216

    • Spark RAPIDS のプロパティ（省略可）: デフォルトでは、Dataproc サーバーレスは次の Spark RAPIDS プロパティ値を設定します。

      • spark.plugins=com.nvidia.spark.SQLPlugin
      • spark.executor.resource.gpu.amount=1
      • spark.task.resource.gpu.amount=1/$spark_executor_cores
      • spark.shuffle.manager=''。デフォルトでは、このプロパティは設定されていません。 ただし、パフォーマンス向上のために GPU を使用する場合には、RAPIDS シャッフル マネージャーを有効にすることが NVIDIA により推奨されています。これを行うには、ワークロードを送信するときに spark.shuffle.manager=com.nvidia.spark.rapids.RapidsShuffleManager を設定します。

      Spark RAPIDS プロパティを設定するには、Apache Spark 構成用の RAPIDS アクセラレータをご覧ください。また、Spark 詳細プロパティを設定するには、Apache Spark 詳細構成用の RAPIDS アクセラレータをご覧ください。