Managed Service for Apache Spark

1. 概要 - Apache Spark 向け Serverless

Managed Service for Apache Spark は、Apache Spark、Apache Flink、Presto をはじめ、他の多くのオープンソース ツールやフレームワークを実行するための、フルマネージドでスケーラビリティの高いサービスです。Managed Apache Spark を使用すると、データレイクのモダナイゼーション、ETL / ELT、安全なデータ サイエンスを世界規模で実現できます。Managed Apache Spark は、BigQuery、Cloud Storage、Gemini Enterprise Agent Engine、Knowledge Catalog など、複数の Google Cloud サービスと完全に統合されています。

マネージド Apache Spark には次の 2 つのバージョンがあります。

  • マネージド Apache Spark サーバーレスを使用すると、インフラストラクチャと自動スケーリングを構成することなく PySpark ジョブを実行できます。マネージド Apache Spark は、PySpark バッチ ワークロードとセッション / ノートブックをサポートしています。
  • マネージド Apache Spark クラスタを使用すると、Flink や Presto などのオープンソース ツールに加えて、YARN ベースの Spark ワークロード用の Hadoop YARN クラスタを管理できます。自動スケーリングなど、必要な垂直スケーリングまたは水平スケーリングの量でクラウドベースのクラスタを調整できます。

この Codelab では、Dataproc Serverless を使用するさまざまな方法について学びます。

当初、Apache Spark は Hadoop クラスタで動作するように構築され、リソース マネージャーとして YARN を使用するものでした。Hadoop クラスタのメンテナンスには、特定の専門知識と、クラスタのさまざまなノブが適切に構成されていることを確認する必要があります。これは、Spark でユーザーが設定する必要がある別のノブのセットに追加されます。そのため、デベロッパーが Spark コード自体ではなくインフラストラクチャの構成に多くの時間を費やしているシナリオが多く発生します。

Dataproc Serverless を使用すると、Hadoop クラスタまたは Spark を手動で構成する必要がなくなります。Dataproc Serverless は Hadoop では実行されず、独自の動的リソース割り当てを使用してリソース要件を判断し、必要に応じて自動スケーリングも行います。Dataproc Serverless を使用すると、Spark プロパティのわずかな部分しかカスタマイズできませんが、ほとんどの場合、これらのプロパティを調整する必要はありません。

2. 設定

まず、この Codelab で使用する環境とリソースを構成します。

Google Cloud プロジェクトを作成します。既存のプロジェクトを使用することもできます。

Cloud コンソールのツールバーでクリックして Cloud Shell を開きます。

ba0bb17945a73543.png

Cloud Shell には、この Codelab で使用できるすぐに使用可能なシェル環境が用意されています。

68c4ebd2a8539764.png

Cloud Shell はデフォルトでプロジェクト名を設定します。echo $GOOGLE_CLOUD_PROJECT を実行して、再度確認します。出力にプロジェクト ID が表示されない場合は、設定します。

export GOOGLE_CLOUD_PROJECT=<your-project-id>

リソースの Compute Engine リージョン（us-central1 や europe-west2 など）を設定します。

export REGION=<your-region>

API を有効にする

この Codelab では、次の API を使用します。

  • BigQuery
  • Dataproc

必要な API を有効にします。これには約 1 分かかります。完了すると、成功メッセージが表示されます。

gcloud services enable bigquery.googleapis.com
gcloud services enable dataproc.googleapis.com

ネットワーク アクセスを構成する

Dataproc Serverless では、Spark ドライバとエグゼキュータにプライベート IP のみが割り当てられているため、Spark ジョブを実行するリージョンで Google プライベート アクセスを有効にする必要があります。次のコマンドを実行して、default サブネットで有効にします。

gcloud compute networks subnets update default \
  --region=${REGION} \
  --enable-private-ip-google-access

Google プライベート アクセスが有効になっているかどうかは、次のコマンドで確認できます。このコマンドは True または False を出力します。

gcloud compute networks subnets describe default \
  --region=${REGION} \
  --format="get(privateIpGoogleAccess)"

Storage バケットを作成する

この Codelab で作成したアセットを保存するために使用するストレージ バケットを作成します。

バケットの名前を選択します。バケット名は、すべてのユーザー間でグローバルに一意である必要があります。

export BUCKET=<your-bucket-name>

Spark ジョブを実行するリージョンにバケットを作成します。

gsutil mb -l ${REGION} gs://${BUCKET}

バケットが Cloud Storage コンソールで使用可能になっていることを確認できます。gsutil ls を実行してバケットを確認することもできます。

永続履歴サーバーを作成する

Spark UI には、デバッグツールと Spark ジョブに関する分析情報が豊富に用意されています。完了した Dataproc サーバーレス ジョブの Spark UI を表示するには、永続履歴サーバーとして使用する単一ノードの Dataproc クラスタを作成する必要があります。

永続履歴サーバーの名前を設定します。

PHS_CLUSTER_NAME=my-phs

次のコマンドを実行します。

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

Spark UI と永続履歴サーバーについては、この Codelab で後ほど詳しく説明します。

3. Dataproc バッチでサーバーレス Spark ジョブを実行する

このサンプルでは、ニューヨーク市（NYC）Citi Bike の移動に関する一般公開データセットの一連のデータを使用します。NYC Citi Bikes は、ニューヨーク市内の有料の自転車シェアリング システムです。簡単な変換を実行し、最も人気のある Citi Bike ステーション ID の上位 10 件を出力します。このサンプルでは、オープンソースの spark-bigquery-connector を使用して、Spark と BigQuery の間でデータをシームレスに読み書きします。

次の Github リポジトリのクローンを作成し、ファイル citibike.py を含むディレクトリに cd します。

git clone https://github.com/GoogleCloudPlatform/devrel-demos.git
cd devrel-demos/data-analytics/next-2022-workshop/dataproc-serverless

citibike.py

import sys

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import BooleanType

if len(sys.argv) == 1:
    print("Please provide a GCS bucket name.")

bucket = sys.argv[1]
table = "bigquery-public-data:new_york_citibike.citibike_trips"

spark = SparkSession.builder \
          .appName("pyspark-example") \
          .config("spark.jars","gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar") \
          .getOrCreate()

df = spark.read.format("bigquery").load(table)

top_ten = df.filter(col("start_station_id") \
            .isNotNull()) \
            .groupBy("start_station_id") \
            .count() \
            .orderBy("count", ascending=False) \
            .limit(10) \
            .cache()

top_ten.show()

top_ten.write.option("header", True).csv(f"gs://{bucket}/citibikes_top_ten_start_station_ids")

デフォルトで Cloud Shell で使用可能な Cloud SDK を使用して、ジョブをサーバーレス Spark に送信します。Cloud SDK と Dataproc Batches API を使用してサーバーレス Spark ジョブを送信する次のコマンドをシェルで実行します。

gcloud dataproc batches submit pyspark citibike.py \
  --batch=citibike-job \
  --region=${REGION} \
  --deps-bucket=gs://${BUCKET} \
  --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar \
--history-server-cluster=projects/${GOOGLE_CLOUD_PROJECT}/regions/${REGION}/clusters/${PHS_CLUSTER_NAME} \
  -- ${BUCKET}

内訳は次のとおりです。

  • gcloud dataproc batches submit は、Dataproc Batches API を参照します。
  • pyspark は、PySpark ジョブを送信することを示します。
  • --batch はジョブの名前です。指定しない場合は、ランダムに生成された UUID が使用されます。
  • --region=${REGION} は、ジョブが処理される地理的リージョンです。
  • --deps-bucket=${BUCKET} は、サーバーレス環境で実行する前にローカル Python ファイルがアップロードされる場所です。
  • --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar には、Spark ランタイム環境の spark-bigquery-connector の jar が含まれています。
  • --history-server-cluster=projects/${GOOGLE_CLOUD_PROJECT}/regions/${REGION}/clusters/${PHS_CLUSTER} は、永続的履歴サーバーの完全修飾名です。ここでは、Spark イベントデータ（コンソール出力とは別）が保存され、Spark UI から表示できます。
  • 末尾の -- は、これ以降のものがプログラムのランタイム引数であることを示します。この場合、ジョブで必要なバケットの名前を送信します。

バッチを送信すると、次の出力が表示されます。

Batch [citibike-job] submitted.

数分後、ジョブのメタデータとともに次の出力が表示されます。

+----------------+------+
|start_station_id| count|
+----------------+------+
|             519|551078|
|             497|423334|
|             435|403795|
|             426|384116|
|             293|372255|
|             402|367194|
|             285|344546|
|             490|330378|
|             151|318700|
|             477|311403|
+----------------+------+

Batch [citibike-job] finished.

次のセクションでは、このジョブのログを見つける方法について説明します。

その他の機能

Spark Serverless を使用すると、ジョブの実行オプションが増えます。

  • ジョブが実行されるカスタム Docker イメージを作成できます。これは、Python や R ライブラリなどの追加の依存関係を含めるのに最適な方法です。
  • Dataproc Metastore インスタンスをジョブに接続して、Hive メタデータにアクセスできます。
  • より細かく制御するために、Dataproc Serverless では、少数の Spark プロパティの構成がサポートされています。

4. Dataproc の指標とオブザーバビリティ

Dataproc バッチ コンソールには、すべての Dataproc Serverless ジョブが一覧表示されます。コンソールには、各ジョブの バッチ ID、ロケーション、ステータス、作成時間、経過時間、タイプが表示されます。ジョブの [バッチ ID] をクリックすると、詳細情報が表示されます。

このページには、[モニタリング] など、ジョブで使用されたバッチ Spark エグゼキュータの数（自動スケーリングの量を示す）が表示されます。

[詳細] タブには、ジョブとともに送信された引数やパラメータなど、ジョブに関するメタデータが表示されます。

このページからすべてのログにアクセスすることもできます。Dataproc Serverless ジョブが実行されると、次の 3 種類のログが生成されます。

  • サービスレベル
  • コンソール出力
  • Spark のイベント ロギング

サービスレベル: Dataproc Serverless サービスによって生成されたログが含まれます。たとえば、Dataproc Serverless が自動スケーリング用に CPU の追加をリクエストする場合などです。これらのログを表示するには、[ログを表示] をクリックします。これにより、Cloud Logging が開きます。

[出力] で [コンソール出力] を確認できます。これは、ジョブによって生成される出力です。これには、ジョブの開始時に Spark が出力するメタデータや、ジョブに組み込まれた print ステートメントが含まれます。

Spark イベント ロギングには、Spark UI からアクセスできます。Spark ジョブに永続履歴サーバーを指定したため、[Spark History Server を表示] をクリックして Spark UI にアクセスできます。この UI には、以前に実行した Spark ジョブの情報が含まれています。Spark UI の詳細については、公式の Spark ドキュメントをご覧ください。

5. Dataproc テンプレート: BQ -> GCS

Dataproc テンプレートは、クラウド内のデータ処理タスクをさらに簡素化するのに役立つオープンソース ツールです。これらは Dataproc Serverless のラッパーとして機能し、次のような多くのデータ インポートおよびエクスポート タスクのテンプレートが含まれています。

  • BigQuerytoGCS、GCStoBigQuery
  • GCStoBigTable
  • GCStoJDBC と JDBCtoGCS
  • HivetoBigQuery
  • MongotoGCS、GCStoMongo

完全なリストは README で確認できます。

このセクションでは、Dataproc テンプレートを使用して BigQuery から GCS にデータをエクスポートします。

リポジトリのクローンを作成する

リポジトリのクローンを作成し、python フォルダに移動します。

git clone https://github.com/GoogleCloudPlatform/dataproc-templates.git
cd dataproc-templates/python

環境を構成する

環境変数を設定します。Dataproc テンプレートでは、プロジェクト ID の環境変数 GCP_PROJECT が使用されるため、これを GOOGLE_CLOUD_PROJECT. に設定します。

export GCP_PROJECT=${GOOGLE_CLOUD_PROJECT}

リージョンは、前の環境で設定されているはずです。設定されていない場合は、ここで設定します。

export REGION=<region>

Dataproc テンプレートは、BigQuery ジョブの処理に spark-bigquery-conector を使用します。このため、URI を環境変数 JARS に含める必要があります。JARS 変数を設定します。

export JARS="gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar"

テンプレート パラメータを構成する

サービスで使用するステージング バケットの名前を設定します。

export GCS_STAGING_LOCATION=gs://${BUCKET}

次に、ジョブ固有の変数を設定します。入力テーブルには、BigQuery NYC Citibike データセットを再度参照します。

BIGQUERY_GCS_INPUT_TABLE=bigquery-public-data.new_york_citibike.citibike_trips

csv、parquet、avro、json のいずれかを選択できます。この Codelab では、CSV を選択します。次のセクションでは、Dataproc テンプレートを使用してファイル形式を変換する方法について説明します。

BIGQUERY_GCS_OUTPUT_FORMAT=csv

出力モードを overwrite に設定します。overwrite、append、ignore、errorifexists. のいずれかを選択できます。

BIGQUERY_GCS_OUTPUT_MODE=overwrite

GCS 出力ロケーションをバケット内のパスに設定します。

BIGQUERY_GCS_OUTPUT_LOCATION=gs://${BUCKET}/BQtoGCS

テンプレートを実行する

次のテンプレートを指定し、設定した入力パラメータを指定して、BIGQUERYTOGCS テンプレートを実行します。

./bin/start.sh \
-- --template=BIGQUERYTOGCS \
        --bigquery.gcs.input.table=${BIGQUERY_GCS_INPUT_TABLE} \
        --bigquery.gcs.output.format=${BIGQUERY_GCS_OUTPUT_FORMAT} \
        --bigquery.gcs.output.mode=${BIGQUERY_GCS_OUTPUT_MODE} \
        --bigquery.gcs.output.location=${BIGQUERY_GCS_OUTPUT_LOCATION}

出力はかなりノイズが多いですが、約 1 分後に次のようになります。

Batch [5766411d6c78444cb5e80f305308d8f8] submitted.
...
Batch [5766411d6c78444cb5e80f305308d8f8] finished.

次のコマンドを実行して、ファイルが生成されたことを確認できます。

gsutil ls ${BIGQUERY_GCS_OUTPUT_LOCATION}

Spark はデフォルトで、データ量に応じて複数のファイルに書き込みます。この場合、生成されるファイルは約 30 個になります。Spark の出力ファイル名は part- の後に 5 桁の数字（パーティション番号を示す）とハッシュ文字列が続いた形式になります。データ量が大きい場合、Spark は通常複数のファイルに書き込みます。ファイル名の例は part-00000-cbf69737-867d-41cc-8a33-6521a725f7a0-c000.csv です。

6. Dataproc テンプレート: CSV から Parquet

ここでは、Dataproc テンプレートを使用して、GCSTOGCS を使用して GCS のデータをあるファイル形式から別のファイル形式に変換します。このテンプレートは SparkSQL を使用し、変換中に処理される SparkSQL クエリを送信して追加の処理を行うオプションも提供します。

環境変数を確認する

GCP_PROJECT、REGION、GCS_STAGING_BUCKET が前のセクションで設定されていることを確認します。

echo ${GCP_PROJECT}
echo ${REGION}
echo ${GCS_STAGING_LOCATION}

テンプレート パラメータを設定する

次に、GCStoGCS の構成パラメータを設定します。入力ファイルの場所から始めます。これはディレクトリであり、特定のファイルではありません。ディレクトリ内のすべてのファイルが処理されます。BIGQUERY_GCS_OUTPUT_LOCATION に設定します。

GCS_TO_GCS_INPUT_LOCATION=${BIGQUERY_GCS_OUTPUT_LOCATION}

入力ファイルの形式を設定します。

GCS_TO_GCS_INPUT_FORMAT=csv

目的の出力形式を設定します。parquet、json、avro、csv を選択できます。

GCS_TO_GCS_OUTPUT_FORMAT=parquet

出力モードを overwrite に設定します。overwrite、append、ignore、errorifexists. のいずれかを選択できます。

GCS_TO_GCS_OUTPUT_MODE=overwrite

出力場所を設定します。

GCS_TO_GCS_OUTPUT_LOCATION=gs://${BUCKET}/GCStoGCS

テンプレートを実行する

GCStoGCS テンプレートを実行します。

./bin/start.sh \
-- --template=GCSTOGCS \
        --gcs.to.gcs.input.location=${GCS_TO_GCS_INPUT_LOCATION} \
        --gcs.to.gcs.input.format=${GCS_TO_GCS_INPUT_FORMAT} \
        --gcs.to.gcs.output.format=${GCS_TO_GCS_OUTPUT_FORMAT} \
        --gcs.to.gcs.output.mode=${GCS_TO_GCS_OUTPUT_MODE} \
        --gcs.to.gcs.output.location=${GCS_TO_GCS_OUTPUT_LOCATION}

出力はかなりノイズが多いですが、約 1 分後に次のような成功メッセージが表示されます。

Batch [c198787ba8e94abc87e2a0778c05ec8a] submitted.
...
Batch [c198787ba8e94abc87e2a0778c05ec8a] finished.

次のコマンドを実行して、ファイルが生成されたことを確認できます。

gsutil ls ${GCS_TO_GCS_OUTPUT_LOCATION}

このテンプレートでは、gcs.to.gcs.temp.view.name と gcs.to.gcs.sql.query をテンプレートに渡すことで SparkSQL クエリを指定することもできます。これにより、GCS に書き込む前にデータに対して SparkSQL クエリを実行できます。

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

この Codelab の完了後に GCP アカウントに不要な料金が発生しないようにするには:

  1. 作成した環境の Cloud Storage バケットを削除します。
gsutil rm -r gs://${BUCKET}
  1. 永続履歴サーバーに使用した Dataproc クラスタを削除します。
gcloud dataproc clusters delete ${PHS_CLUSTER_NAME} \
  --region=${REGION}
  1. Dataproc Serverless ジョブを削除します。Batches コンソールに移動し、削除するジョブの横にあるチェックボックスをオンにして、[削除] をクリックします。

この Codelab 専用のプロジェクトを作成した場合は、必要に応じてプロジェクトを削除することもできます。

  1. GCP Console で [プロジェクト] ページに移動します。
  2. プロジェクト リストで、削除するプロジェクトを選択し、[削除] をクリックします。
  3. ボックスにプロジェクト ID を入力し、[シャットダウン] をクリックしてプロジェクトを削除します。

8. 次のステップ

次のリソースでは、サーバーレス Spark を活用するその他の方法について説明します。