搭配使用 BigQuery 連接器和 Spark

您可以搭配 Apache Spark 使用 BigQuery 連接器,在 BigQuery 中讀取及寫入資料。本頁提供搭配 Spark 使用 BigQuery 連接器的程式碼範例。如需建立叢集的操作說明,請參閱 Cloud Dataproc 快速入門

在 BigQuery 中讀取和寫入資料

此範例會使用 SparkContext.newAPIHadoopRDD,將 BigQuery 中的資料讀取至 Spark 以計算字數 (詳情請參閱 Spark 說明文件),然後使用 PairRDDFunctions.saveAsNewAPIHadoopDataset 將資料寫回 BigQuery。

嘗試執行此範例之前,請先建立資料集並命名為「wordcount_dataset」,然後在程式碼中將輸出資料集變更為 Google Cloud Platform 專案中的現有 BigQuery 資料集。下面是建立資料集的 bq 指令:

bq mk wordcount_dataset

Scala

檢查程式碼

IndirectBigQueryOutputFormat 提供 Hadoop 且可讓您將 JsonObject 值直接寫入 BigQuery 表格。這個類別可透過 Hadoop OutputFormat 類別的擴充功能,提供 BigQuery 記錄的存取權。如要正確使用,必須在 Hadoop 設定中設定數個參數,且 OutputFormat 類別必須設為 IndirectBigQueryOutputFormat。下列範例說明要設定的參數,以及正確使用 IndirectBigQueryOutputFormat 所需的幾行程式碼。
import com.google.cloud.hadoop.io.bigquery.BigQueryConfiguration
import com.google.cloud.hadoop.io.bigquery.BigQueryFileFormat
import com.google.cloud.hadoop.io.bigquery.GsonBigQueryInputFormat
import com.google.cloud.hadoop.io.bigquery.output.BigQueryOutputConfiguration
import com.google.cloud.hadoop.io.bigquery.output.IndirectBigQueryOutputFormat
import com.google.gson.JsonObject
import org.apache.hadoop.io.LongWritable
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat

// Assumes you have a spark context (sc) -- running from spark-shell REPL.
// Marked as transient since configuration is not Serializable. This should
// only be necessary in spark-shell REPL.
@transient
val conf = sc.hadoopConfiguration

// Input parameters.
val fullyQualifiedInputTableId = "publicdata:samples.shakespeare"
val projectId = conf.get("fs.gs.project.id")
val bucket = conf.get("fs.gs.system.bucket")

// Input configuration.
conf.set(BigQueryConfiguration.PROJECT_ID_KEY, projectId)
conf.set(BigQueryConfiguration.GCS_BUCKET_KEY, bucket)
BigQueryConfiguration.configureBigQueryInput(conf, fullyQualifiedInputTableId)

// Output parameters.
val outputTableId = projectId + ":wordcount_dataset.wordcount_output"
// Temp output bucket that is deleted upon completion of job.
val outputGcsPath = ("gs://" + bucket + "/hadoop/tmp/bigquery/wordcountoutput")

// Output configuration.
// Let BigQuery auto-detect output schema (set to null below).
BigQueryOutputConfiguration.configureWithAutoSchema(
    conf,
    outputTableId,
    outputGcsPath,
    BigQueryFileFormat.NEWLINE_DELIMITED_JSON,
    classOf[TextOutputFormat[_,_]])

// (Optional) Configure the KMS key used to encrypt the output table.
BigQueryOutputConfiguration.setKmsKeyName(
    conf, "projects/myproject/locations/us-west1/keyRings/r1/cryptoKeys/k1");

conf.set("mapreduce.job.outputformat.class",
         classOf[IndirectBigQueryOutputFormat[_,_]].getName)

// Truncate the table before writing output to allow multiple runs.
conf.set(BigQueryConfiguration.OUTPUT_TABLE_WRITE_DISPOSITION_KEY,
         "WRITE_TRUNCATE")

// Helper to convert JsonObjects to (word, count) tuples.
def convertToTuple(record: JsonObject) : (String, Long) = {
  val word = record.get("word").getAsString.toLowerCase
  val count = record.get("word_count").getAsLong
  return (word, count)
}

// Helper to convert (word, count) tuples to JsonObjects.
def convertToJson(pair: (String, Long)) : JsonObject = {
  val word = pair._1
  val count = pair._2
  val jsonObject = new JsonObject()
  jsonObject.addProperty("word", word)
  jsonObject.addProperty("word_count", count)
  return jsonObject
}

// Load data from BigQuery.
val tableData = sc.newAPIHadoopRDD(
    conf,
    classOf[GsonBigQueryInputFormat],
    classOf[LongWritable],
    classOf[JsonObject])

// Perform word count.
val wordCounts = (tableData
    .map(entry => convertToTuple(entry._2))
    .reduceByKey(_ + _))

// Display 10 results.
wordCounts.take(10).foreach(l => println(l))

// Write data back into a new BigQuery table.
// IndirectBigQueryOutputFormat discards keys, so set key to null.
(wordCounts
    .map(pair => (null, convertToJson(pair)))
    .saveAsNewAPIHadoopDataset(conf))

在叢集中執行程式碼

  1. 使用 SSH 連結 Cloud Dataproc 叢集的主要節點
    1. 前往 GCP 主控台中專案的 Cloud Dataproc「Clusters」(叢集) 頁面,然後按一下叢集名稱
    2. 在叢集詳細資料頁面,選取 [VM Instances] (VM 執行個體) 分頁標籤,然後按一下叢集主要節點名稱右邊的 [SSH] 選項

      瀏覽器視窗會開啟主要節點上的主目錄
          Connected, host fingerprint: ssh-rsa 2048 ...
          ...
          user@clusterName-m:~$
          
  2. 使用預先安裝的 vivimnano 文字編輯器建立 wordcount.scala,然後貼上從 Scala 程式碼清單複製而來的 Scala 程式碼
    nano wordcount.scala
      
  3. 啟動 spark-shell REPL (如果是使用 Cloud Dataproc 1.0-1.2,則移除 jars 參數)。
    $ spark-shell --jars=gs://hadoop-lib/bigquery/bigquery-connector-hadoop2-latest.jar
    ...
    Using Scala version ...
    Type in expressions to have them evaluated.
    Type :help for more information.
    ...
    Spark context available as sc.
    ...
    SQL context available as sqlContext.
    scala>
    
  4. 使用 :load wordcount.scala 指令執行 wordcount.scala,以建立 BigQuery wordcount_output 表格。輸出清單會顯示 wordcount 輸出中的 10 行內容。
    :load wordcount.scala
    ...
    (pinnace,3)
    (bone,21)
    (lug,2)
    (vailing,2)
    (bombast,3)
    (gaping,11)
    (hem,5)
    ('non,1)
    (stinks,1)
    (forsooth,48)
    

    如要預覽輸出資料表,請開啟專案的 BigQuery 頁面,選取 wordcount_output 資料表,然後按一下 [Preview] (預覽)

PySpark

檢查程式碼

#!/usr/bin/python
"""BigQuery I/O PySpark example."""
from __future__ import absolute_import
import json
import pprint
import subprocess
import pyspark
from pyspark.sql import SQLContext

sc = pyspark.SparkContext()

# Use the Cloud Storage bucket for temporary BigQuery export data used
# by the InputFormat. This assumes the Cloud Storage connector for
# Hadoop is configured.
bucket = sc._jsc.hadoopConfiguration().get('fs.gs.system.bucket')
project = sc._jsc.hadoopConfiguration().get('fs.gs.project.id')
input_directory = 'gs://{}/hadoop/tmp/bigquery/pyspark_input'.format(bucket)

conf = {
    # Input Parameters.
    'mapred.bq.project.id': project,
    'mapred.bq.gcs.bucket': bucket,
    'mapred.bq.temp.gcs.path': input_directory,
    'mapred.bq.input.project.id': 'publicdata',
    'mapred.bq.input.dataset.id': 'samples',
    'mapred.bq.input.table.id': 'shakespeare',
}

# Output Parameters.
output_dataset = 'wordcount_dataset'
output_table = 'wordcount_output'

# Load data in from BigQuery.
table_data = sc.newAPIHadoopRDD(
    'com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat',
    'org.apache.hadoop.io.LongWritable',
    'com.google.gson.JsonObject',
    conf=conf)

# Perform word count.
word_counts = (
    table_data
    .map(lambda record: json.loads(record[1]))
    .map(lambda x: (x['word'].lower(), int(x['word_count'])))
    .reduceByKey(lambda x, y: x + y))

# Display 10 results.
pprint.pprint(word_counts.take(10))

# Stage data formatted as newline-delimited JSON in Cloud Storage.
output_directory = 'gs://{}/hadoop/tmp/bigquery/pyspark_output'.format(bucket)
output_files = output_directory + '/part-*'

sql_context = SQLContext(sc)
(word_counts
 .toDF(['word', 'word_count'])
 .write.format('json').save(output_directory))

# Shell out to bq CLI to perform BigQuery import.
subprocess.check_call(
    'bq load --source_format NEWLINE_DELIMITED_JSON '
    '--replace '
    '--autodetect '
    '{dataset}.{table} {files}'.format(
        dataset=output_dataset, table=output_table, files=output_files
    ).split())

# Manually clean up the staging_directories, otherwise BigQuery
# files will remain indefinitely.
input_path = sc._jvm.org.apache.hadoop.fs.Path(input_directory)
input_path.getFileSystem(sc._jsc.hadoopConfiguration()).delete(input_path, True)
output_path = sc._jvm.org.apache.hadoop.fs.Path(output_directory)
output_path.getFileSystem(sc._jsc.hadoopConfiguration()).delete(
    output_path, True)

在叢集中執行程式碼

  1. 使用 SSH 連結 Cloud Dataproc 叢集的主要節點
    1. 前往 GCP 主控台中專案的 Cloud Dataproc「Clusters」(叢集) 頁面,然後按一下叢集名稱
    2. 在叢集詳細資料頁面,選取 [VM Instances] (VM 執行個體) 分頁標籤,然後按一下叢集主要節點名稱右邊的 [SSH] 選項

      瀏覽器視窗會開啟主要節點上的主目錄
          Connected, host fingerprint: ssh-rsa 2048 ...
          ...
          user@clusterName-m:~$
          
  2. 使用預先安裝的 vivimnano 文字編輯器建立 wordcount.py,然後貼上從 PySpark 程式碼清單複製而來的 PySpark 程式碼
    nano wordcount.py
      
  3. 使用 spark-submit 執行 wordcount,以建立 BigQuery wordcount_output 表格。輸出清單會顯示 wordcount 輸出中的 10 行內容。
    spark-submit wordcount.py
    ...
    (pinnace,3)
    (bone,21)
    (lug,2)
    (vailing,2)
    (bombast,3)
    (gaping,11)
    (hem,5)
    ('non,1)
    (stinks,1)
    (forsooth,48)
    

    如要預覽輸出資料表,請開啟專案的 BigQuery 頁面,選取 wordcount_output 資料表,然後按一下 [Preview] (預覽)