Apache Spark SQL connector for Google BigQuery

The connector supports reading Google BigQuery tables into Spark's DataFrames, and writing DataFrames back into BigQuery. This is done by using the Spark SQL Data Source API to communicate with BigQuery.

Unreleased Changes

This Readme may include documentation for changes that haven't been released yet. The latest release's documentation and source code are found here.

https://github.com/GoogleCloudDataproc/spark-bigquery-connector/blob/master/README.md

BigQuery Storage API

The Storage API streams data in parallel directly from BigQuery via gRPC without using Google Cloud Storage as an intermediary.

It has a number of advantages over using the previous export-based read flow that should generally lead to better read performance:

Direct Streaming

It does not leave any temporary files in Google Cloud Storage. Rows are read directly from BigQuery servers using the Arrow or Avro wire formats.

Filtering

The new API allows column and predicate filtering to only read the data you are interested in.

Column Filtering

Since BigQuery is backed by a columnar datastore, it can efficiently stream data without reading all columns.

Predicate Filtering

The Storage API supports arbitrary pushdown of predicate filters. Connector version 0.8.0-beta and above support pushdown of arbitrary filters to Bigquery.

There is a known issue in Spark that does not allow pushdown of filters on nested fields. For example - filters like address.city = "Sunnyvale" will not get pushdown to Bigquery.

Dynamic Sharding

The API rebalances records between readers until they all complete. This means that all Map phases will finish nearly concurrently. See this blog article on how dynamic sharding is similarly used in Google Cloud Dataflow.

See Configuring Partitioning for more details.

Requirements

Enable the BigQuery Storage API

Follow these instructions.

Create a Google Cloud Dataproc cluster (Optional)

If you do not have an Apache Spark environment you can create a Cloud Dataproc cluster with pre-configured auth. The following examples assume you are using Cloud Dataproc, but you can use spark-submit on any cluster.

Any Dataproc cluster using the API needs the 'bigquery' or 'cloud-platform' scopes. Dataproc clusters have the 'bigquery' scope by default, so most clusters in enabled projects should work by default e.g.

MY_CLUSTER=...
gcloud dataproc clusters create "$MY_CLUSTER"

Downloading and Using the Connector

The latest version of the connector is publicly available in the following links:

The first six versions are Java based connectors targeting Spark 2.4/3.1/3.2/3.3/3.4/3.5 of all Scala versions built on the new Data Source APIs (Data Source API v2) of Spark.

The final two connectors are Scala based connectors, please use the jar relevant to your Spark installation as outlined below.

Connector to Spark Compatibility Matrix

Connector to Dataproc Image Compatibility Matrix

| Connector \ Dataproc Image | 1.3 | 1.4 | 1.5 | 2.0 | 2.1 | 2.2 | Serverless

Image 1.0 | Serverless

Image 2.0 | Serverless

Image 2.1 | Serverless

Image 2.2 | |---------------------------------------|---------|---------|---------|---------|---------|---------|-------------------------|-------------------------|-------------------------|-------------------------| | spark-3.5-bigquery | | | | | | ✓ | | | | ✓ | | spark-3.4-bigquery | | | | | | ✓ | | | ✓ | ✓ | | spark-3.3-bigquery | | | | | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | | spark-3.2-bigquery | | | | | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | | spark-3.1-bigquery | | | | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | | spark-2.4-bigquery | | ✓ | ✓ | | | | | | | | | spark-bigquery-with-dependencies_2.13 | | | | | | | | ✓ | ✓ | ✓ | | spark-bigquery-with-dependencies_2.12 | | | ✓ | ✓ | ✓ | ✓ | ✓ | | | | | spark-bigquery-with-dependencies_2.11 | ✓ | ✓ | | | | | | | | |

Maven / Ivy Package Usage

The connector is also available from the Maven Central repository. It can be used using the --packages option or the spark.jars.packages configuration property. Use the following value

Specifying the Spark BigQuery connector version in a Dataproc cluster

Dataproc clusters created using image 2.1 and above, or batches using the Dataproc serverless service come with built-in Spark BigQuery connector. Using the standard --jars or --packages (or alternatively, the spark.jars/spark.jars.packages configuration) won't help in this case as the built-in connector takes precedence.

To use another version than the built-in one, please do one of the following:

• For Dataproc clusters, using image 2.1 and above, add the following flag on cluster creation to upgrade the version --metadata SPARK_BQ_CONNECTOR_VERSION=${next-release-tag}, or --metadata SPARK_BQ_CONNECTOR_URL=gs://spark-lib/bigquery/spark-3.3-bigquery-${next-release-tag}.jar to create the cluster with a different jar. The URL can point to any valid connector JAR for the cluster's Spark version.
• For Dataproc serverless batches, add the following property on batch creation to upgrade the version: --properties dataproc.sparkBqConnector.version=${next-release-tag}, or --properties dataproc.sparkBqConnector.uri=gs://spark-lib/bigquery/spark-3.3-bigquery-${next-release-tag}.jar to create the batch with a different jar. The URL can point to any valid connector JAR for the runtime's Spark version.

Hello World Example

You can run a simple PySpark wordcount against the API without compilation by running

Dataproc image 1.5 and above

gcloud dataproc jobs submit pyspark --cluster "$MY_CLUSTER" \
  --jars gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-${next-release-tag}.jar \
  examples/python/shakespeare.py

Dataproc image 1.4 and below

gcloud dataproc jobs submit pyspark --cluster "$MY_CLUSTER" \
  --jars gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.11-0.29.0.jar \
  examples/python/shakespeare.py

Example Codelab

https://codelabs.developers.google.com/codelabs/pyspark-bigquery

Usage

The connector uses the cross language Spark SQL Data Source API:

Reading data from a BigQuery table

``` df = spark.read \ .format(