Apache Spark 用 Google Cloud Serverless で GPU を使用する

GPU アクセラレータを Google Cloud Apache Spark 向けサーバーレスバッチワークロードにアタッチすると、次の結果が得られます。

大規模なデータ分析ワークロードの処理を高速化します。
GPU 機械学習ライブラリを使用して、大規模なデータセットでのモデルトレーニングを高速化します。
動画や自然言語処理などの高度なデータ分析を実行します。

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

GPU のメリット

Apache Spark 向け Serverless Spark ワークロードで GPU を使用する場合のメリットは次のとおりです。

パフォーマンスの向上: GPU アクセラレーションにより、Spark ワークロードのパフォーマンスを大幅に向上させることができます。特に、機械学習やディープラーニング、グラフ処理、複雑な分析など、コンピューティング負荷の高いタスクで効果を発揮します。
モデルトレーニングの高速化: 機械学習タスクの場合、GPU を接続すると、モデルのトレーニングに必要な時間を大幅に短縮できるため、データサイエンティストやエンジニアは迅速に反復処理と実験を行うことができます。
スケーラビリティ: 複雑化する処理ニーズに対応するため、GPU ノードを追加したり、より強力な GPU をノードに追加したりできます。
費用対効果: GPU には初期投資が必要ですが、処理時間の短縮とリソース使用率の向上により、長期的にコストを削減できます。
データ分析の強化: GPU アクセラレーションにより、画像分析、動画分析、自然言語処理などの高度な分析を大規模なデータセットで実行できます。
プロダクトの改善: 処理の高速化により、意思決定を迅速に行い、アプリケーションの応答性を高めることができます。

制限事項と考慮事項

GPU アクセラレータは、プレミアム料金階層で利用できます。
NVIDIA A100 または NVIDIA L4 の GPU を Google Cloud Apache Spark 向けサーバーレスバッチワークロードにアタッチできます。A100 および L4 アクセラレータは、 Compute Engine GPU を利用できるリージョンの対象です。
XGBoost ライブラリは、Apache Spark 向け Serverless GPU 高速化ワークロードに、Apache Spark 向け Serverless Spark ランタイムバージョン 1.x を使用しているときにのみ提供されます。
XGBoost を使用した Apache Spark 向け Serverless GPU 高速化バッチでは、増加した Compute Engine 割り当てが使用されます。たとえば、NVIDIA L4 GPU を使用するサーバーレスバッチワークロードを実行するには、 NVIDIA_L4_GPUS 割り当てを割り当てる必要があります。
アクセラレータが有効になっているジョブには、 constraints/compute.requireShieldedVm 組織のポリシーとの互換性がありません。組織でこのポリシーが適用されると、アクセラレータが有効になっているジョブは正常に実行されません。
サポートされている Apache Spark 向け Serverless ランタイムで RAPIDS GPU アクセラレーションを使用する場合は、バージョン 2.2より前のバージョンでデフォルトの文字セットを UTF-8 に設定する必要があります。詳細については、 GPU アクセラレータを使用してサーバーレスバッチワークロードを作成するをご覧ください。

料金

GPU アクセラレータは、プレミアム料金階層で利用できます。アクセラレータの料金については、Apache Spark 向け Serverless の料金をご覧ください。

始める前に

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

アカウントにログインします。 Google Cloud を初めて使用する場合は、アカウントを作成して、実際のシナリオで Google プロダクトのパフォーマンスを評価してください。 Google Cloud新規のお客様には、ワークロードの実行、テスト、デプロイができる無料クレジット $300 分を差し上げます。

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

Roles required to select or create a project

Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

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

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

Google Cloud CLI をインストールします。

外部 ID プロバイダ（IdP）を使用している場合は、まず連携 ID を使用して gcloud CLI にログインする必要があります。

gcloud CLI を初期化するには、次のコマンドを実行します:

gcloud init

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

Roles required to select or create a project

Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

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

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

Google Cloud CLI をインストールします。

外部 ID プロバイダ（IdP）を使用している場合は、まず連携 ID を使用して gcloud CLI にログインする必要があります。

gcloud CLI を初期化するには、次のコマンドを実行します:

gcloud init

コンソールで Cloud Storage の Google Cloud [**バケット**] ページに移動します。
[バケット] に移動
[ [Create]] をクリックします。
[バケットの作成] ページでユーザーのバケット情報を入力します。次のステップに進むには、[続行] をクリックします。
1. [スタートガイド] セクションで、次の操作を行います。
  - バケット名の要件を満たす、グローバルに一意の名前を入力します。
  - バケットラベルを追加するには、[ラベル] セクション（）を開き、[ラベルを追加] をクリックして、key と value を指定します。
2. [データの保存場所の選択] セクションで、次の操作を行います。
  1. ロケーションタイプを選択してください。
  2. [Location type] プルダウンメニューから、バケットのデータが永続的に保存されるロケーションを選択します。
    - ロケーションタイプとして [デュアルリージョン] を選択した場合は、関連するチェックボックスを使用して [ターボレプリケーション] を有効にすることもできます。
  3. クロスバケットレプリケーションを設定するには、 [Storage Transfer Service 経由でクロスバケットレプリケーションを追加する] を選択し、次の手順を実施します:
    クロスバケットレプリケーションを設定する
    
    [バケット] メニューで、バケットを選択します。
    
    [レプリケーション設定] セクションで [構成] をクリックして、レプリケーションジョブの設定を構成します。
    
    [**クロスバケットレプリケーションを構成する**] ペインが表示されます。
    
    オブジェクト名の接頭辞で複製するオブジェクトをフィルタするには、オブジェクトを追加または除外する接頭辞を入力し、 [接頭辞を追加] をクリックします。
    
    複製されたオブジェクトのストレージクラスを設定するには、 [Storage class] メニューからストレージクラスを選択します。この手順をスキップすると、複製されたオブジェクトはデフォルトで宛先バケットのストレージクラスを使用します。
    
    [完了] をクリックします。
3. [データの保存場所の選択] セクションで、次の操作を行います。
  1. バケットのデフォルトのストレージクラスを選択するか、バケットデータのストレージクラスを自動的に管理するAutoclassを選択します。
  2. 階層名前空間を有効にするには、 [データ量が多いワークロード向けにストレージを最適化] セクションで、 [このバケットで階層的な名前空間を有効にする] を選択します。
    注: 既存のバケットで階層名前空間を有効にすることはできません。
4. In the [オブジェクトへのアクセスを制御する方法を選択する] セクションで、バケットに公開アクセスの防止を適用するかどうかを選択し、バケットのオブジェクトに使用するアクセス制御方法を選択します。
  **注:** 組織の組織ポリシーでこの設定が適用されている場合、[**公開アクセスの防止**] 設定を変更することはできません。
5. [オブジェクトデータを保護する方法を選択する] セクションで、次の操作を行います。
  - [**データ保護**] で、バケットに設定するオプションを選択します。
    - 削除（復元可能）を有効にするには、 [削除（復元可能）ポリシー（データ復旧用）] チェックボックスをオンにして、削除後にオブジェクトを保持する日数を指定します。
    - オブジェクトのバージョニングを設定するには、 [オブジェクトのバージョニング（バージョン管理用）] チェックボックスをオンにして、オブジェクトあたりの最大バージョン数と、非現行バージョンが期限切れになるまでの日数を指定します。
    - オブジェクトとバケットで保持ポリシーを有効にするには、[保持（コンプライアンス用）] チェックボックスをオンにして、次の操作を行います。
      - [オブジェクト保持ロック]を有効にするには、 [オブジェクト保持を有効にする]チェックボックスをオンにします。
      - [Bucket Lock] を有効にするには、[バケット保持ポリシーを設定する] チェックボックスをオンにして、保持期間の単位と保持期間を選択します。
  - オブジェクトデータの暗号化方法を選択するには、 [データ暗号化] セクション（）を開き、 [データの暗号化] 方法を選択します。
[作成] をクリックします。

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

NVIDIA L4 GPU を使用する Apache Spark 向け Serverless バッチワークロードを送信して、並列化された PySpark タスクを実行します。gcloud CLI を使用して次の手順を行います。

[展開する] をクリックし、テキストまたはコードエディタを使用して、一覧表示された 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()
)

ローカルマシンで gcloud CLI を使用して、各ワーカーを L4 GPU で高速化した 5 ワーカーで、Apache Spark 向け 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: サポートされているすべての Supported Google Cloud Apache Spark 向け Serverless ランタイムで、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 高速化ワークロードは、プレミアム データコンピューティングユニット（DCU）を使用して課金されます。Apache Spark 向け Serverless アクセラレータの料金をご覧ください。
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 です。

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

spark.executor.memory（省略可）と spark.executor.memoryOverhead（制限あり）: メモリは設定できますが、 memoryOverhead は設定できません。設定されたプロパティで使用されていない使用可能なメモリの量は、設定されていないプロパティに適用されます。 spark.executor.memoryOverhead は PySpark バッチワークロードの使用可能なメモリの 40% に設定され、他のワークロードの場合は 10% に設定されます（ Spark のリソース割り当てプロパティをご覧ください）。

次の表に、A100 GPU と 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 のプロパティ（省略可）: デフォルトでは、Apache Spark 向け Serverless は次の 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.sql.concurrentGpuTasks= minimum of (gpuMemoryinMB / 8, 4)
- spark.rapids.shuffle.multiThreaded.writer.threads= minimum of (cpu cores in the VM / gpu count per VM, 32)
- spark.rapids.shuffle.multiThreaded.reader.threads= minimum of (cpu cores in the VM / gpu count per VM, 32)
Spark RAPIDS プロパティを設定するには、Apache Spark 構成用の RAPIDS アクセラレータをご覧ください。また、Spark 詳細プロパティを設定するには、Apache Spark 詳細構成用の RAPIDS アクセ/3} ラレータをご覧ください。

Apache Spark 用 Google Cloud Serverless で GPU を使用する コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。

GPU のメリット

制限事項と考慮事項

料金

始める前に

クロスバケット レプリケーションを設定する

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

Apache Spark 用 Google Cloud Serverless で GPU を使用する

クロスバケットレプリケーションを設定する

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