Use the BigQuery connector with Spark

/*
 * Remove comment if you are not running in spark-shell.
 *
import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder()
  .appName("spark-bigquery-demo")
  .getOrCreate()
*/

// Use the Cloud Storage bucket for temporary BigQuery export data used
// by the connector.
val bucket = "[bucket]"
spark.conf.set("temporaryGcsBucket", bucket)

// Load data in from BigQuery. See
// https://github.com/GoogleCloudDataproc/spark-bigquery-connector/tree/0.17.3#properties
// for option information.
val wordsDF =
  (spark.read.format("bigquery")
  .option("table","bigquery-public-data:samples.shakespeare")
  .load()
  .cache())

wordsDF.createOrReplaceTempView("words")

// Perform word count.
val wordCountDF = spark.sql(
  "SELECT word, SUM(word_count) AS word_count FROM words GROUP BY word")
wordCountDF.show()
wordCountDF.printSchema()

// Saving the data to BigQuery.
(wordCountDF.write.format("bigquery")
  .option("table","wordcount_dataset.wordcount_output")
  .save())

1. Use SSH to connect to the Dataproc cluster master node
   1. Go to the Dataproc Clusters page in the Google Cloud console, then click the name of your cluster

      

   2. On the >Cluster details page, select the VM Instances tab. Then, click SSH to the right of the name of the cluster master node

      

      
      A browser window opens at your home directory on the master node

          Connected, host fingerprint: ssh-rsa 2048 ...
          ...
          user@clusterName-m:~$
          

2. Create wordcount.scala with the pre-installed vi, vim, or nano text editor, then paste in the Scala code from the Scala code listing

   nano wordcount.scala
     

3. Launch the spark-shell REPL.

   $ spark-shell --jars=gs://spark-lib/bigquery/spark-bigquery-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. Run wordcount.scala with the :load wordcount.scala command to create the BigQuery wordcount_output table. The output listing displays 20 lines from the wordcount output.

   :load wordcount.scala
   ...
   +---------+----------+
   |     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)
   

   
   To preview the output table, open the BigQuery page, select the wordcount_output table, and then click Preview.

   

#!/usr/bin/env python

"""BigQuery I/O PySpark example."""

from pyspark.sql import SparkSession

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

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

# Load data from BigQuery.
words = spark.read.format('bigquery') \
  .option('table', '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') \
  .option('table', 'wordcount_dataset.wordcount_output') \
  .save()

1. Use SSH to connect to the Dataproc cluster master node
   1. Go to the Dataproc Clusters page in the Google Cloud console, then click the name of your cluster

      

   2. On the Cluster details page, select the VM Instances tab. Then, click SSH to the right of the name of the cluster master node

      

      
      A browser window opens at your home directory on the master node

          Connected, host fingerprint: ssh-rsa 2048 ...
          ...
          user@clusterName-m:~$
          

2. Create wordcount.py with the pre-installed vi, vim, or nano text editor, then paste in the PySpark code from the PySpark code listing

   nano wordcount.py
   

3. Run wordcount with spark-submit to create the BigQuery wordcount_output table. The output listing displays 20 lines from the wordcount output.

   spark-submit --jars gs://spark-lib/bigquery/spark-bigquery-latest.jar wordcount.py
   ...
   +---------+----------+
   |     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)
   

   
   To preview the output table, open the BigQuery page, select the wordcount_output table, and then click Preview.