Repository files navigation

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

TheStorage 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:

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.

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

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

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 likeaddress.city = "Sunnyvale" will not get pushdown to Bigquery.

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

SeeConfiguring Partitioning for more details.

Followthese instructions.

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 usespark-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"

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 newData 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 outlinedbelow.

The connector is also available from theMaven Centralrepository. It can be used using the--packages option or thespark.jars.packages configuration property. Use the following value

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, thespark.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=0.42.2, or--metadata SPARK_BQ_CONNECTOR_URL=gs://spark-lib/bigquery/spark-3.3-bigquery-0.42.2.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=0.42.2, or--properties dataproc.sparkBqConnector.uri=gs://spark-lib/bigquery/spark-3.3-bigquery-0.42.2.jar to create the batch with a different jar. The URL can point to any valid connector JAR for the runtime's Spark version.

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-0.42.2.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

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

The connector uses the cross languageSpark SQL Data Source API:

df = spark.read \  .format("bigquery") \  .load("bigquery-public-data.samples.shakespeare")

or the Scala only implicit API:

import com.google.cloud.spark.bigquery._val df = spark.read.bigquery("bigquery-public-data.samples.shakespeare")

For more information, see additional code samples inPython,ScalaandJava.

The connector allows you to run anyStandard SQLSELECT query on BigQuery and fetch its results directly to a Spark Dataframe.This is easily done as described in the following code sample:

spark.conf.set("viewsEnabled","true")sql = """  SELECT tag, COUNT(*) c  FROM (    SELECT SPLIT(tags, '|') tags    FROM `bigquery-public-data.stackoverflow.posts_questions` a    WHERE EXTRACT(YEAR FROM creation_date)>=2014  ), UNNEST(tags) tag  GROUP BY 1  ORDER BY 2 DESC  LIMIT 10  """df = spark.read.format("bigquery").load(sql)df.show()

Which yields the result

+----------+-------+|       tag|      c|+----------+-------+|javascript|1643617||    python|1352904||      java|1218220||   android| 913638||       php| 911806||        c#| 905331||      html| 769499||    jquery| 608071||       css| 510343||       c++| 458938|+----------+-------+

A second option is to use thequery option like this:

df = spark.read.format("bigquery").option("query", sql).load()

Notice that the execution should be faster as only the result is transmittedover the wire. In a similar fashion the queries can include JOINs moreefficiently then running joins on Spark or use other BigQuery features such assubqueries,BigQuery user defined functions,wildcard tables,BigQuery MLand more.

In order to use this feature theviewsEnabled configurations MUST be set totrue. This can also be done globally as shown in the example above.

Important: This feature is implemented by running the query on BigQuery andsaving the result into a temporary table, of which Spark will read the resultsfrom. This may add additional costs on your BigQuery account.

The connector supports executingBigQuery parameterized queries using thestandardspark.read.format('bigquery') API.

To use parameterized queries:

1. Provide the SQL query containing parameters using the.option("query", "SQL_STRING") with named (@param) or positional (?) parameters.
2. Specify the parameter values using dedicated options:

• Named Parameters: Use options prefixed withNamedParameters.. Theparameter name follows the prefix (case-insensitive).
  • Format:.option("NamedParameters.<parameter_name>", "TYPE:value")
  • Example:.option("NamedParameters.corpus", "STRING:romeoandjuliet")
• Positional Parameters: Use options prefixed withPositionalParameters.. The 1-based index follows the prefix.
  • Format:.option("PositionalParameters.<parameter_index>", "TYPE:value")
  • Example:.option("PositionalParameters.1", "STRING:romeoandjuliet")

TheTYPE in theTYPE:value string specifies the BigQuery Standard SQL datatype. Supported types currently include:BOOL,INT64,FLOAT64,NUMERIC,STRING,DATE,DATETIME,JSON,TIME,GEOGRAPHY,TIMESTAMP.

ARRAY andSTRUCT types are not supported as parameters at this time.

The connector has a preliminary support for reading fromBigQuery views. Pleasenote there are a few caveats:

• BigQuery views are not materialized by default, which means that the connectorneeds to materialize them before it can read them. This process affects theread performance, even before running anycollect() orcount() action.
• The materialization process can also incur additional costs to your BigQuerybill.
• Reading from views isdisabled by default. In order to enable it,either set the viewsEnabled option when reading the specific view(.option("viewsEnabled", "true")) or set it globally by callingspark.conf.set("viewsEnabled", "true").

Notice: Before version 0.42.1 of the connector, the following configurationsare required:

• By default, the materialized views are created in the same project anddataset. Those can be configured by the optionalmaterializationProjectandmaterializationDataset options, respectively. These options can alsobe globally set by callingspark.conf.set(...) before reading the views.
• As mentioned in theBigQuery documentation,thematerializationDataset should be in same location as the view.

Starting version 0.42.1 those configurations areredundant and are ignored.It is highly recommended to upgrade to this version or a later one to enjoysimpler configuration when using views or loading from queries.

Writing DataFrames to BigQuery can be done using two methods: Direct and Indirect.

In this method the data is written directly to BigQuery using theBigQuery Storage Write API. In order to enable this option, pleaseset thewriteMethod option todirect, as shown below:

df.write \  .format("bigquery") \  .option("writeMethod", "direct") \  .option("writeAtLeastOnce", "true")  .save("dataset.table")

Writing to existing partitioned tables (date partitioned, ingestion time partitioned and rangepartitioned) in APPEND save mode and OVERWRITE mode (only date and range partitioned) is fully supported by the connector and the BigQuery Storage WriteAPI. The use ofdatePartition,partitionField,partitionType,partitionRangeStart,partitionRangeEnd,partitionRangeIntervaldescribed below is not supported at this moment by the direct write method.

Important: Please refer to thedata ingestion pricingpage regarding the BigQuery Storage Write API pricing.

Important: Please use version 0.24.2 and above for direct writes, as previousversions have a bug that may cause a table deletion in certain cases.

In this method the data is written first to GCS, and then it is loaded it to BigQuery. A GCS bucket must be configuredto indicate the temporary data location.

df.write \  .format("bigquery") \  .option("temporaryGcsBucket","some-bucket") \  .save("dataset.table")

The data is temporarily stored using theApache Parquet,Apache ORC orApache Avro formats.

The GCS bucket and the format can also be set globally using Spark's RuntimeConfig like this:

spark.conf.set("temporaryGcsBucket","some-bucket")df.write \  .format("bigquery") \  .save("dataset.table")

When streaming a DataFrame to BigQuery, each batch is written in the same manner as a non-streaming DataFrame.Note that a HDFS compatiblecheckpoint location(eg:path/to/HDFS/dir orgs://checkpoint-bucket/checkpointDir) must be specified.

df.writeStream \  .format("bigquery") \  .option("temporaryGcsBucket","some-bucket") \  .option("checkpointLocation", "some-location") \  .option("table", "dataset.table")

Important: The connector does not configure the GCS connector, in order to avoid conflict with another GCS connector, if exists. In order to use the write capabilities of the connector, please configure the GCS connector on your cluster as explainedhere.

The API Supports a number of options to configure the read

Options can also be set outside of the code, using the--conf parameter ofspark-submit or--properties parameterof thegcloud dataproc submit spark. In order to use this, prepend the prefixspark.datasource.bigquery. to any ofthe options, for examplespark.conf.set("temporaryGcsBucket", "some-bucket") can also be set as--conf spark.datasource.bigquery.temporaryGcsBucket=some-bucket.

With the exception ofDATETIME andTIME all BigQuery data types directed map into the corresponding Spark SQL data type. Here are all of the mappings:

The Spark MLVector andMatrix are supported,including their dense and sparse versions. The data is saved as a BigQuery RECORD. Notice that a suffix is added tothe field's description which includes the spark type of the field.

In order to write those types to BigQuery, use the ORC or Avro intermediate format, and have them as column of theRow (i.e. not a field in a struct).

BigQuery's BigNumeric has a precision of 76.76 (the 77th digit is partial) and scale of 38. Sincethis precision and scale is beyond spark's DecimalType (38 scale and 38 precision) support, it meansthat BigNumeric fields with precision larger than 38 cannot be used. Once this Spark limitation willbe updated the connector will be updated accordingly.

The Spark Decimal/BigQuery Numeric conversion tries to preserve the parameterization of the type, i.eNUMERIC(10,2) will be converted toDecimal(10,2) and vice versa. Notice however that there arecases wherethe parameters are lost.This means that the parameters will be reverted to the defaults - NUMERIC (38,9) and BIGNUMERIC(76,38).This means that at the moment, BigNumeric read is supported only from a standard table, but not fromBigQuery view or whenreading data from a BigQuery query.

The connector automatically computes column and pushdown filters the DataFrame'sSELECT statement e.g.

spark.read.bigquery("bigquery-public-data:samples.shakespeare")  .select("word")  .where("word = 'Hamlet' or word = 'Claudius'")  .collect()

filters to the columnword and pushed down the predicate filterword = 'hamlet' or word = 'Claudius'.

If you do not wish to make multiple read requests to BigQuery, you can cache the DataFrame before filtering e.g.:

val cachedDF = spark.read.bigquery("bigquery-public-data:samples.shakespeare").cache()val rows = cachedDF.select("word")  .where("word = 'Hamlet'")  .collect()// All of the table was cached and this doesn't require an API callval otherRows = cachedDF.select("word_count")  .where("word = 'Romeo'")  .collect()

You can also manually specify thefilter option, which will override automatic pushdown and Spark will do the rest of the filtering in the client.

The pseudo columns _PARTITIONDATE and _PARTITIONTIME are not part of the table schema. Therefore in order to query by the partitions ofpartitioned tables do not use the where() method shown above. Instead, add a filter option in the following manner:

val df = spark.read.format("bigquery")  .option("filter", "_PARTITIONDATE > '2019-01-01'")  ...  .load(TABLE)

By default the connector creates one partition per 400MB in the table being read (before filtering). This should roughly correspond to the maximum number of readers supported by the BigQuery Storage API.This can be configured explicitly with themaxParallelism property. BigQuery may limit the number of partitions based on server constraints.

In order to support tracking the usage of BigQuery resources the connectorsoffers the following options to tag BigQuery resources:

The connector can launch BigQuery load and query jobs. Adding labels to the jobsis done in the following manner:

spark.conf.set("bigQueryJobLabel.cost_center", "analytics")spark.conf.set("bigQueryJobLabel.usage", "nightly_etl")

This will create labelscost_center=analytics andusage=nightly_etl.

Used to annotate the read and write sessions. The trace ID is of the formatSpark:ApplicationName:JobID. This is an opt-in option, and to use it the userneed to set thetraceApplicationName property. JobID is auto generated by theDataproc job ID, with a fallback to the Spark application ID (such asapplication_1648082975639_0001). The Job ID can be overridden by setting thetraceJobId option. Notice that the total length of the trace ID cannot be over256 characters.

The connector can be used inJupyter notebooks even ifit is not installed on the Spark cluster. It can be added as an external jar inusing the following code:

Python:

frompyspark.sqlimportSparkSessionspark=SparkSession.builder \  .config("spark.jars.packages","com.google.cloud.spark:spark-bigquery-with-dependencies_2.12:0.42.2") \  .getOrCreate()df=spark.read.format("bigquery") \  .load("dataset.table")

Scala:

valspark=SparkSession.builder.config("spark.jars.packages","com.google.cloud.spark:spark-bigquery-with-dependencies_2.12:0.42.2").getOrCreate()valdf= spark.read.format("bigquery").load("dataset.table")

In case Spark cluster is using Scala 2.12 (it's optional for Spark 2.4.x,mandatory in 3.0.x), then the relevant package iscom.google.cloud.spark:spark-bigquery-with-dependencies_2.12:0.42.2. Inorder to know which Scala version is used, please run the following code:

Python:

spark.sparkContext._jvm.scala.util.Properties.versionString()

Scala:

scala.util.Properties.versionString

Unless you wish to use the implicit Scala APIspark.read.bigquery("TABLE_ID"), there is no need to compile against the connector.

To include the connector in your project:

<dependency>  <groupId>com.google.cloud.spark</groupId>  <artifactId>spark-bigquery-with-dependencies_${scala.version}</artifactId>  <version>0.42.2</version></dependency>

libraryDependencies+="com.google.cloud.spark"%%"spark-bigquery-with-dependencies"%"0.42.2"

Spark populates a lot of metrics which can be found by the end user in the spark history page. But all these metrics are spark related which are implicitly collected without any change from the connector.But there are few metrics which are populated from the BigQuery and currently are visible in the application logs which can be read in the driver/executor logs.

From Spark 3.2 onwards, spark has provided the API to expose custom metrics in the spark UI pagehttps://spark.apache.org/docs/3.2.0/api/java/org/apache/spark/sql/connector/metric/CustomMetric.html

Currently, using this API, connector exposes the following bigquery metrics during read

Note: To use the metrics in the Spark UI page, you need to make sure thespark-bigquery-metrics-0.42.2.jar is the class path before starting the history-server and the connector version isspark-3.2 or above.

See theBigQuery pricing documentation.

You can manually set the number of partitions with themaxParallelism property. BigQuery may provide fewer partitions than you ask for. SeeConfiguring Partitioning.

You can also always repartition after reading in Spark.

If there are too many partitions the CreateWriteStream or Throughputquotasmay be exceeded. This occurs because while the data within each partition is processed serially, independentpartitions may be processed in parallel on different nodes within the spark cluster. Generally, to ensure maximumsustained throughput you should file a quota increase request. However, you can also manually reduce the number ofpartitions being written by callingcoalesce on the DataFrame to mitigate this problem.

desiredPartitionCount = 5dfNew = df.coalesce(desiredPartitionCount)dfNew.write

A rule of thumb is to have a single partition handle at least 1GB of data.

Also note that a job running with thewriteAtLeastOnce property turned on will not encounter CreateWriteStreamquota errors.

The connector needs an instance of a GoogleCredentials in order to connect to the BigQuery APIs. There are multipleoptions to provide it:

• The default is to load the JSON key from theGOOGLE_APPLICATION_CREDENTIALS environment variable, as describedhere.
• In case the environment variable cannot be changed, the credentials file can be configured asas a spark option. The file should reside on the same path on all the nodes of the cluster.

// Globallyspark.conf.set("credentialsFile", "</path/to/key/file>")// Per read/Writespark.read.format("bigquery").option("credentialsFile", "</path/to/key/file>")

• Credentials can also be provided explicitly, either as a parameter or from Spark runtime configuration.They should be passed in as a base64-encoded string directly.

// Globallyspark.conf.set("credentials", "<SERVICE_ACCOUNT_JSON_IN_BASE64>")// Per read/Writespark.read.format("bigquery").option("credentials", "<SERVICE_ACCOUNT_JSON_IN_BASE64>")

• In cases where the user has an internal service providing the Google AccessToken, a custom implementationcan be done, creating only the AccessToken and providing its TTL. Token refresh will re-generate a new token. In orderto use this, implement thecom.google.cloud.bigquery.connector.common.AccessTokenProviderinterface. The fully qualified class name of the implementation should be provided in thegcpAccessTokenProvideroption.AccessTokenProvider must be implemented in Java or other JVM language such as Scala or Kotlin. It musteither have a no-arg constructor or a constructor accepting a singlejava.util.String argument. This configurationparameter can be supplied using thegcpAccessTokenProviderConfig option. If this is not provided then the no-argconstructor wil be called. The jar containing the implementation should be on the cluster's classpath.

// Globallyspark.conf.set("gcpAccessTokenProvider", "com.example.ExampleAccessTokenProvider")// Per read/Writespark.read.format("bigquery").option("gcpAccessTokenProvider", "com.example.ExampleAccessTokenProvider")

• Service account impersonation can be configured for a specific username and a group name, orfor all users by default using below properties:

  • gcpImpersonationServiceAccountForUser_<USER_NAME> (not set by default)

    The service account impersonation for a specific user.

  • gcpImpersonationServiceAccountForGroup_<GROUP_NAME> (not set by default)

    The service account impersonation for a specific group.

  • gcpImpersonationServiceAccount (not set by default)

    Default service account impersonation for all users.

  If any of the above properties are set then the service account specified will be impersonated bygenerating a short-lived credentials when accessing BigQuery.

  If more than one property is set then the service account associated with the username will takeprecedence over the service account associated with the group name for a matching user and group,which in turn will take precedence over default service account impersonation.

• For a simpler application, where access token refresh is not required, another alternative is to pass the access tokenas thegcpAccessToken configuration option. You can get the access token by runninggcloud auth application-default print-access-token.

// Globallyspark.conf.set("gcpAccessToken", "<access-token>")// Per read/Writespark.read.format("bigquery").option("gcpAccessToken", "<acccess-token>")

Important: TheCredentialsProvider andAccessTokenProvider need to be implemented in Java orother JVM language such as Scala or Kotlin. The jar containing the implementation should be on the cluster's classpath.

Notice: Only one of the above options should be provided.

To connect to a forward proxy and to authenticate the user credentials, configure the following options.

proxyAddress: Address of the proxy server. The proxy must be an HTTP proxy and address should be in thehost:portformat.

proxyUsername: The userName used to connect to the proxy.

proxyPassword: The password used to connect to the proxy.

val df = spark.read.format("bigquery")  .option("proxyAddress", "http://my-proxy:1234")  .option("proxyUsername", "my-username")  .option("proxyPassword", "my-password")  .load("some-table")

The same proxy parameters can also be set globally using Spark's RuntimeConfig like this:

spark.conf.set("proxyAddress", "http://my-proxy:1234")spark.conf.set("proxyUsername", "my-username")spark.conf.set("proxyPassword", "my-password")val df = spark.read.format("bigquery")  .load("some-table")

You can set the following in the hadoop configuration as well.

fs.gs.proxy.address(similar to "proxyAddress"),fs.gs.proxy.username(similar to "proxyUsername") andfs.gs.proxy.password(similar to "proxyPassword").

If the same parameter is set at multiple places the order of priority is as follows:

option("key", "value") > spark.conf > hadoop configuration