Spark BigQuery コネクタを使用する

Managed Service for Apache Spark で spark-bigquery-connector を使用すると、BigQuery との間でデータの読み書きを行えます。このチュートリアルでは、spark-bigquery-connector を使用する PySpark アプリケーションについて説明します。

コネクタのバージョンを確認する

バッチ ワークロードまたはインタラクティブ セッションのランタイム バージョンにインストールされている BigQuery コネクタのバージョンを確認するには、Managed Service for Apache Spark ランタイム リリースをご覧ください。コネクタがリストにない場合は、コネクタをアプリケーションで使用可能にするをご覧ください。

必要に応じて、アプリケーションでコネクタを使用できるようにする

BigQuery コネクタは、サポートされている Managed Service for Apache Spark ランタイム バージョンすべてにインストールされます。コネクタ（Spark runtime 1.0）をインストールしないサポートされていないランタイム バージョンを使用している場合は、次のいずれかの方法でコネクタをアプリケーションで使用できます。

  • Managed Service for Apache Spark バッチ ワークロードを送信するか、インタラクティブ セッションを実行するときに、jars パラメータを使用してコネクタ jar ファイルを指定します。次のバッチ ワークロードの例では、コネクタ JAR ファイルを指定しています（使用可能なコネクタ JAR ファイルのリストについては、GitHub の GoogleCloudDataproc/Spark-Bigquery コネクタ リポジトリをご覧ください）。
    • Google Cloud CLI の例: 
      gcloud dataproc batches submit pyspark \
    --region=REGION \
    --jars=spark-3.5-bigquery-version.jar \
    ... other args

費用の計算

このチュートリアルでは、以下を含む、 Google Cloudの課金対象となるコンポーネントを使用します。

  • Managed Service for Apache Spark
  • BigQuery
  • Cloud Storage

料金計算ツールを使うと、予想使用量に基づいて費用の見積もりを作成できます。

Cloud Platform の新規ユーザーは無料トライアルをご利用いただけます。

課金を構成する

デフォルトでは、認証情報またはサービス アカウントに関連付けられているプロジェクトには、API 使用量に対して課金されます。別のプロジェクトに請求するには、次の構成プロパティを設定します。spark.conf.set("parentProject", "<BILLED-GCP-PROJECT>")

また、読み取り / 書き込みオペレーションに次のように追加することもできます。.option("parentProject", "<BILLED-GCP-PROJECT>")

PySpark wordcount バッチ ワークロードを送信する

このサンプルでは、BigQuery から Spark DataFrame にデータを読み込み、標準データソース API を使用してワード数をカウントします。

コネクタは、次のオペレーション シーケンスでワードカウント出力を BigQuery に書き込みます。

  1. Cloud Storage バケット内の一時ファイルにデータをバッファリングします

  2. 1 回のオペレーションで Cloud Storage バケットから BigQuery にデータをコピーします

  3. BigQuery の読み込みオペレーションが完了すると、Cloud Storage 内の一時ファイルが削除されます（一時ファイルは Spark アプリケーションが終了した後にも削除されます）。削除に失敗した場合は、不要な Cloud Storage 一時ファイルを削除する必要があります。ファイルは通常、gs://BUCKET_NAME/.spark-bigquery-JOB_ID-UUID にあります。

wordcount ワークロードを実行する手順

  1. ローカル ターミナルまたは Cloud Shell を開きます。
  2. ローカル ターミナルまたは Cloud Shell で、bq コマンドライン ツールを使用して wordcount_dataset を作成します。
    bq mk wordcount_dataset
  3. Google Cloud CLI を使用して Cloud Storage バケットを作成します。
    gcloud storage buckets create gs://BUCKET_NAME
    BUCKET_NAME は、作成した Cloud Storage バケットの名前に置き換えます。
  4. 次の PySpark コードをコピーして、テキスト エディタでファイル wordcount.py をローカルに作成します。 
    #!/usr/bin/python
"""BigQuery I/O PySpark example."""
from pyspark.sql import SparkSession

spark = SparkSession \
  .builder \
  .appName('spark-bigquery-demo') \
  .getOrCreate()

# Cloud Storage bucket used by the connector for temporary BigQuery
# export data.
bucket = "BUCKET_NAME"
spark.conf.set('temporaryGcsBucket', bucket)

# Load data from BigQuery.
words = spark.read.format('bigquery') \
  .load('bigquery-public-data.samples.shakespeare') \
  .load()
words.createOrReplaceTempView('words')

# Perform word count.
word_count = spark.sql(
    'SELECT word, SUM(word_count) AS word_count FROM words GROUP BY word')
word_count.show()
word_count.printSchema()

# Save the data to BigQuery
word_count.write.format('bigquery') \
  .save('wordcount_dataset.wordcount_output')
  5. PySpark バッチ ワークロードを送信します。
    gcloud dataproc batches submit pyspark wordcount.py \
    --region=REGION \
    --deps-bucket=BUCKET_NAME
    ターミナル出力例: 
    ...
+---------+----------+
|     word|word_count|
+---------+----------+
|     XVII|         2|
|    spoil|        28|
|    Drink|         7|
|forgetful|         5|
|   Cannot|        46|
|    cures|        10|
|   harder|        13|
|  tresses|         3|
|      few|        62|
|  steel'd|         5|
| tripping|         7|
|   travel|        35|
|   ransom|        55|
|     hope|       366|
|       By|       816|
|     some|      1169|
|    those|       508|
|    still|       567|
|      art|       893|
|    feign|        10|
+---------+----------+
only showing top 20 rows

root
 |-- word: string (nullable = false)
 |-- word_count: long (nullable = true)

    Google Cloud コンソールで出力テーブルをプレビューするには、[BigQuery] ページを開き、wordcount_output テーブルを選択して、[プレビュー] をクリックします。
    BigQuery テーブルのプレビューのレンダリング
    図 1: BigQuery で出力テーブルをプレビューする

詳細情報