Apache Spark バッチ ワークロードを実行する

Dataproc Serverless を使用して、必要に応じてリソースをスケールする Dataproc マネージド コンピューティング インフラストラクチャのバッチ ワークロードを送信する方法を学習します。

準備

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

2. Google Cloud Console の [プロジェクト セレクタ] ページで、Google Cloud プロジェクトを選択または作成します。

   注: この手順で作成するリソースをそのまま保持する予定でない場合、既存のプロジェクトを選択するのではなく、新しいプロジェクトを作成してください。チュートリアルの終了後にそのプロジェクトを削除すれば、プロジェクトに関連するすべてのリソースを削除できます。

   プロジェクト セレクタに移動

3. Google Cloud プロジェクトで課金が有効になっていることを確認します。

4. Dataproc API を有効にします。

   API を有効にする

5. Google Cloud Console の [プロジェクト セレクタ] ページで、Google Cloud プロジェクトを選択または作成します。

   注: この手順で作成するリソースをそのまま保持する予定でない場合、既存のプロジェクトを選択するのではなく、新しいプロジェクトを作成してください。チュートリアルの終了後にそのプロジェクトを削除すれば、プロジェクトに関連するすべてのリソースを削除できます。

   プロジェクト セレクタに移動

6. Google Cloud プロジェクトで課金が有効になっていることを確認します。

7. Dataproc API を有効にします。

   API を有効にする

Spark バッチ ワークロードの送信

コンソール

1. Google Cloud コンソールで、Dataproc バッチに移動します。[作成] をクリックして [バッチ作成] ページを開きます。

2. pi の近似値を計算する Spark バッチ ワークロードを送信するには、ページの次のフィールドを選択して入力します。

   • バッチ情報:
     • バッチ ID: バッチ ワークロードの ID を指定します。この値は 4 ～ 63 文字にする必要があります。有効な文字は /[a-z][0-9]-/ です。
     • リージョン: ワークロードが実行されるリージョンを選択します。
   • コンテナ:
     • バッチタイプ: Spark
     • ランタイム バージョン: デフォルトのランタイム バージョンが選択されています。必要に応じて、デフォルト以外の Dataproc サーバーレス ランタイム バージョンを指定できます。
     • メインクラス:

       org.apache.spark.examples.SparkPi

     • JAR ファイル（このファイルは Dataproc Serverless Spark 実行環境にプリインストールされています）。

       file:///usr/lib/spark/examples/jars/spark-examples.jar

     • 引数: 1000。
   • 実行構成: ワークロードの実行に使用するサービス アカウントを指定できます。サービス アカウントを指定しない場合、ワークロードは Compute Engine のデフォルトのサービス アカウントで実行されます。
   • ネットワーク構成: Dataproc Serverless for Spark ワークロードを実行する VPC サブネットワークは、限定公開の Google アクセスが有効になっており、Dataproc Serverless for Spark のネットワーク構成にリストされているその他の要件を満たす必要があります。サブネットワークのリストには、選択したネットワーク内で限定公開の Google アクセスが有効になっているサブネットが表示されます。
   • プロパティ: Spark バッチ ワークロードに設定するサポートされている Spark のプロパティの Key（プロパティ名）と Value を入力します。注: Compute Engine のクラスタ プロパティの Dataproc とは異なり、Dataproc Serverless for Spark のワークロード プロパティに spark: 接頭辞が含まれていません。
   • その他のオプション:
     • 外部のセルフマネージド Hive メタストアを使用するようにバッチ ワークロードを構成できます。
     • 永続履歴サーバー（PHS）を使用できます。PHS は、バッチ ワークロードを実行するリージョンに配置する必要があります。

3. [送信] をクリックして、Spark バッチ ワークロードを実行します。

gcloud

Spark バッチ ワークロードを送信して pi の近似値を計算するには、次の gcloud CLI の gcloud dataproc batches submit spark コマンドをターミナル ウィンドウまたは Cloud Shell でローカルに実行します。

gcloud dataproc batches submit spark \
    --region=REGION \
    --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
    --class=org.apache.spark.examples.SparkPi \
    -- 1000

注:

• REGION: ワークロードが実行されるリージョンを指定します。
• サブネットワーク: Dataproc Serverless for Spark ワークロードを実行する VPC サブネットワークは、限定公開の Google アクセスが有効になっており、Dataproc Serverless for Spark のネットワーク構成にリストされているその他の要件を満たす必要があります。 gcloud dataproc batches submit コマンドで指定されたリージョンの default ネットワークのサブネットが限定公開の Google アクセスで有効になっていない場合、次のいずれかを行う必要があります。
  • 限定公開の Google アクセスでリージョンのデフォルト ネットワークのサブネットを有効にする
  • 限定公開の Google アクセスが有効になっているサブネットを指定するには、コマンドで --subnet=[SUBNET_URI] フラグを使用します。ネットワーク内のサブネットの URI を一覧表示するには、gcloud compute networks describe [NETWORK_NAME] コマンドを実行します。
• --jars: サンプル JAR ファイルが Spark 実行環境にプリインストールされています。SparkPi ワークロードに渡される 1000 コマンド引数は、円周率の見積もりロジックを 1,000 回繰り返し指定します。（ワークロード入力引数は「--」の後に含まれます）
• --properties: --properties フラグを追加して、Spark バッチ ワークロードで使用するサポートされている Spark プロパティを入力できます。
• --deps-bucket: このフラグを追加して、Dataproc サーバーレスがワークロードの依存関係をアップロードする Cloud Storage バケットを指定できます。バケットの gs:// URI 接頭辞は必要ありません。バケットパスまたはバケット名を指定できます（例: "mybucketname"）。Dataproc Serverless for Spark は、バッチ ワークロードを実行する前に、ローカル ファイルをバケット内の /dependencies フォルダにアップロードします。注: バッチ ワークロードがローカルマシン上のファイルを参照する場合、このフラグは必須です。
• --ttl: --ttl フラグを追加して、バッチの存続期間を指定できます。ワークロードがこの時間を超えると、進行中の作業の完了を待たずにワークロードが無条件に終了します。期間は、s、m、h、d（秒、分、時、日）の接尾辞を使用して指定します。最小値は 10 分（10m）、最大値は 14 日（14d）です。
  • 1.1 または 2.0 ランタイム バッチ: 1.1 または 2.0 ランタイム バッチ ワークロードで --ttl が指定されていない場合、ワークロードは自然に終了するまで実行されます（終了しない場合、永続的に実行されます）。
  • 2.1 以降のランタイム バッチ: 2.1 以降のランタイム バッチ ワークロードで --ttl が指定されていない場合、デフォルトは 4h です。
• その他のオプション: gcloud dataproc batches submit コマンドフラグを追加して、他のワークロード オプションと Spark プロパティを指定できます。
  • Hive メタストア: 次のコマンドは、標準の Spark 構成を使用する外部のセルフマネージド Hive メタストアを使用するようにバッチ ワークロードを構成します。

    gcloud dataproc batches submit \
        --properties=spark.sql.catalogImplementation=hive,spark.hive.metastore.uris=METASTORE_URI,spark.hive.metastore.warehouse.dir=WAREHOUSE_DIR> \
        other args ...
            

  • 永続的履歴サーバー:
    1. 次のコマンドは、単一ノードの Dataproc クラスタに PHS を作成します。PHS は、バッチ ワークロードを実行するリージョンに配置する必要があり、Cloud Storage bucket-name が存在している必要があります。

       gcloud dataproc clusters create PHS_CLUSTER_NAME \
           --region=REGION \
           --single-node \
           --enable-component-gateway \
           --properties=spark:spark.history.fs.logDirectory=gs://bucket-name/phs/*/spark-job-history
                    

    2. 実行中の永続履歴サーバーを指定してバッチ ワークロードを送信します。

       gcloud dataproc batches submit spark \
           --region=REGION \
           --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
           --class=org.apache.spark.examples.SparkPi \
           --history-server-cluster=projects/project-id/regions/region/clusters/PHS-cluster-name \
           -- 1000
                     

  • ランタイム バージョン: --version フラグを使用して、ワークロードの Dataproc サーバーレス ランタイム バージョンを指定します。

    gcloud dataproc batches submit spark \
        --region=REGION \
        --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
        --class=org.apache.spark.examples.SparkPi \
        --version=VERSION
        -- 1000
                

API

このセクションでは、Dataproc Serverless for Spark の batches.create を使用して、pi の近似値を計算するバッチ ワークロードを作成する方法について説明します。

リクエストのデータを使用する前に、次のように置き換えます。

• project-id: Google Cloud プロジェクト ID。
• region: Dataproc サーバーレスがワークロードを実行する Compute Engine リージョン。
注:
• Custom-container-image: Docker イメージの命名形式 {hostname}/{project-id}/{image}:{tag} を使用して、カスタム コンテナ イメージを指定します。例: gcr.io/my-project-id/my-image:1.0.1"。注: カスタム コンテナは Container Registry でホストする必要があります。
• サブネットワーク: Dataproc Serverless for Spark ワークロードを実行する VPC サブネットワークは、限定公開の Google アクセスが有効になっており、Dataproc Serverless for Spark のネットワーク構成にリストされているその他の要件を満たす必要があります。 指定されたリージョンの default ネットワークのサブネットが限定公開の Google アクセスで有効になっていない場合、次のいずれかを行う必要があります。
  1. 限定公開の Google アクセスでリージョンのデフォルト ネットワークのサブネットを有効にする
  2. ExecutionConfig.subnetworkUri フィールドを使用して、限定公開の Google アクセスが有効になっているサブネットを指定します。ネットワーク内のサブネットの URI を一覧表示するには、gcloud compute networks describe [NETWORK_NAME] コマンドを実行します。
• sparkBatch.jarFileUris: サンプルの jar ファイルは、Spark 実行環境にプリインストールされています。「1000」の sparkBatch.args が SparkPi ワークロードに渡されると、pi の見積もりロジックを 1000 回反復するように指定します。
• Spark properties: RuntimeConfig.properties フィールドを使用して、Spark バッチ ワークロードで使用するサポートされている Spark プロパティを入力できます。
• --ttl: EnvironmentConfig.ttl フィールドを使用して、バッチの存続期間を指定できます。ワークロードがこの時間を超えると、進行中の作業の完了を待たずにワークロードが無条件に終了します。 [期間] で期間を JSON 表現として指定します。 最小値は 10 分、最大値は 14 日です。
  • 1.1 または 2.0 ランタイム バッチ: 1.1 または 2.0 ランタイム バッチ ワークロードで --ttl が指定されていない場合、ワークロードは自然に終了するまで実行されます（終了しない場合、永続的に実行されます）。
  • 2.1 以降のランタイム バッチ: 2.1 以降のランタイム バッチ ワークロードで --ttl が指定されていない場合、デフォルトは 4 時間です。
• その他の方法:
  • 外部のセルフマネージド Hive メタストアを使用するようにバッチ ワークロードを構成します。
  • 永続履歴サーバー（PHS）を使用します。 PHS は、バッチ ワークロードを実行するリージョンに配置する必要があります。
  • batches.create リクエストの一部として RuntimeConfig.version フィールドを使用して、デフォルト以外の Dataproc サーバーレス ランタイム バージョンを指定します。

HTTP メソッドと URL:

POST https://dataproc.googleapis.com/v1/projects/project-id/locations/region/batches

リクエストの本文（JSON）:

{
  "sparkBatch":{
    "args":[
      "1000"
    ],
    "jarFileUris":[
      "file:///usr/lib/spark/examples/jars/spark-examples.jar"
    ],
    "mainClass":"org.apache.spark.examples.SparkPi"
  }
}

リクエストを送信するには、次のいずれかのオプションを展開します。

curl（Linux、macOS、Cloud Shell）

注: 次のコマンドは、gcloud init または gcloud auth login を実行して、ユーザー アカウントで gcloud CLI にログインしているか、Cloud Shell を使用して自動的に gcloud CLI にログインしていることを前提としています。gcloud auth list を実行すると、現在アクティブなアカウントを確認できます。

リクエスト本文を request.json という名前のファイルに保存して、次のコマンドを実行します。

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://dataproc.googleapis.com/v1/projects/project-id/locations/region/batches"

PowerShell（Windows）

注:次のコマンドは、gcloud init または gcloud auth login を実行してユーザー アカウントで gcloud CLI にログインしていることを前提としています。gcloud auth list を実行すると、現在アクティブなアカウントを確認できます。

リクエスト本文を request.json という名前のファイルに保存して、次のコマンドを実行します。

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://dataproc.googleapis.com/v1/projects/project-id/locations/region/batches" | Select-Object -Expand Content

次のような JSON レスポンスが返されます。

{
"name":"projects/project-id/locations/region/batches/batch-id",
  "uuid":",uuid",
  "createTime":"2021-07-22T17:03:46.393957Z",
  "sparkBatch":{
    "mainClass":"org.apache.spark.examples.SparkPi",
    "args":[
      "1000"
    ],
    "jarFileUris":[
      "file:///usr/lib/spark/examples/jars/spark-examples.jar"
    ]
  },
  "runtimeInfo":{
    "outputUri":"gs://dataproc-.../driveroutput"
  },
  "state":"SUCCEEDED",
  "stateTime":"2021-07-22T17:06:30.301789Z",
  "creator":"account-email-address",
  "runtimeConfig":{
    "properties":{
      "spark:spark.executor.instances":"2",
      "spark:spark.driver.cores":"2",
      "spark:spark.executor.cores":"2",
      "spark:spark.app.name":"projects/project-id/locations/region/batches/batch-id"
    }
  },
  "environmentConfig":{
    "peripheralsConfig":{
      "sparkHistoryServerConfig":{
      }
    }
  },
  "operation":"projects/project-id/regions/region/operation-id"
}

ワークロードによる円周率の計算方法: Spark ワークロードは、モンテカルロ法を使用して円周率の値を推定します。これによって、単位正方形で囲まれた円をモデル化した座標平面上に x、y 点が生成されます。入力引数（'1000'）は、生成する x と y のペア数を決定します。生成するペアが多いほど、推定の精度が向上します。詳細については、モンテカルロ法を使用した円周率の推定と、GitHub の JavaSparkPi.java をご覧ください。

ワークロードの費用を見積もる

Dataproc Serverless for Spark ワークロードは、データ コンピューティング ユニット（DCU）とシャッフル ストレージ リソースを消費します。Dataproc UsageMetrics を出力してワークロードのリソース消費量と費用を見積もる例については、Dataproc サーバーレスの料金をご覧ください。

次のステップ

以下の内容をご確認いただけます。

• Dataproc サーバーレスの割り当て
• Spark プロパティ