# Google BigQuery Sink Connector for Confluent Platform (Legacy) The Kafka Connect Google BigQuery Sink connector is used to stream data into BigQuery tables. When streaming data from Apache Kafka® topics that have registered schemas, the sink connector can create BigQuery tables with the appropriate BigQuery table schema. The BigQuery table schema is based upon information in the Kafka schema for the topic. Confluent Cloud is available through [Google Cloud Marketplace](https://console.cloud.google.com/marketplace/product/confluent-prod/apache-kafka-on-confluent-cloud?inv=1&invt=Ab2Ryw) or [directly from Confluent](https://www.confluent.io/get-started/). It is important to note the following: - The BigQuery Sink V2 connector is not supported in Confluent Platform. It is only [supported in Confluent Cloud](/cloud/current/connectors/cc-gcp-bigquery-storage-sink.html). - Version 2.x.x of the Google BigQuery Sink connector for Confluent Platform is not backward compatible with 1.x.x versions. For more information, see the [Upgrading to 2.x.x](#upgrading-bigquery-2-0-x) section. - Google BigQuery Sink connector for Confluent Platform (legacy) will remain supported until stated otherwise and should not be considered as End of Support (EoS). - If you want to use multiple connectors with a high number of tasks, ensure you review [BigQuery rate limits](https://cloud.google.com/bigquery/quotas#streaming_inserts). #### IMPORTANT Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the [Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8) page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and are no longer available for download. ## Features - [At least once delivery](#bigquery-sink-at-least-once-delivery) - [Dead Letter Queue](#bigquery-sink-dead-letter-queue) - [Multiple tasks](#bigquery-sink-multiple-tasks) - [Insert operations and duplicate detection](#bigquery-sink-insert-operations) - [Insertion of records one at a time](#bigquery-sink-record-insertion) - [Streaming from a list of topics](#bigquery-sink-streaming-list) - [Internal thread pool](#bigquery-sink-internal-thread-pool) - [Client-side encryption](#bigquery-sink-csfle-sm) ### At least once delivery This connector guarantees that records are delivered at least once from the Kafka topic. ### Dead Letter Queue This connector supports routing invalid records to the [Dead Letter Queue (DLQ)](/platform/current/connect/concepts.html#dead-letter-queue). This includes any records having a `400` code (invalid error message) from BigQuery. Note that DLQ routing does not work if `allowSchemaUnionization` is set to `false` and `allowNewBigQueryFields` and `allowBigQueryRequiredFieldRelaxation` are set to `true` (which is equivalent to setting `autoUpdateSchemas` to `true` in [versions earlier than 2.0.0](#upgrading-bigquery-2-0-x) of this connector) and the connector detects that the failure is due to schema mismatch. For information about accessing and using the DLQ, see [Confluent Platform Dead Letter Queue](/platform/current/connect/concepts.html#dead-letter-queue). #### NOTE There are cases when DLQ feature may require configuring `bigQueryRetry` to a non-zero value. The connector may send a batch of records in one API call. If the API identifies any records within the batch as INVALID, it rejects the entire batch. As a result, even the valid records within the batch are not processed or ingested. The retries to BigQuery API in all cases is controlled by `bigQueryRetry` configuration in the connector. By setting this parameter to a positive value, the connector is instructed to retry sending the batch, thus ensuring that valid records are not discarded. ### Multiple tasks The Google BigQuery Sink connector supports running one or more tasks. You can specify the number of tasks in the `tasks.max` configuration parameter. This can lead to performance gains when multiple files need to be parsed. ### Insert operations and duplicate detection The [InsertAll API](https://cloud.google.com/bigquery/docs/streaming-data-into-bigquery) supports insert operations and tries to detect duplicates. For more information, see [BigQuery troubleshooting](https://cloud.google.com/bigquery/streaming-data-into-bigquery#troubleshooting). ### Insertion of records one at a time The connector uses the BigQuery [insertAll streaming api](https://googleapis.dev/java/google-cloud-clients/latest/com/google/cloud/bigquery/InsertAllRequest.html) which inserts records which are available immediately in the table for querying. ### Streaming from a list of topics The connector supports streaming from a list of topics into corresponding tables in BigQuery. ### Internal thread pool Even though the BigQuery connector streams records one at a time by default (as opposed to running in batch mode), the connector is scalable because it contains an internal thread pool that allows it to stream records in parallel. Note that the internal thread pool defaults to 10 threads, which is configurable. ### Client-side encryption This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html). ## Limitations The BigQuery Sink connector has the following limitations: * The connector does not support schemas with recursion. * The connector does not support schemas having float fields with NaN or +Infinity values. * Auto schema update does not support removing columns. * Auto schema update does not support recursive schemas. - When the connector is configured with `upsertEnabled` or `deleteEnabled`, it does not support [Single Message Transformations (SMTs)](https://docs.confluent.io/platform/current/connect/transforms/overview.html) that modify the topic name. Additionally, the following transformations are not allowed: - `io.debezium.transforms.ByLogicalTableRouter` - `io.debezium.transforms.outbox.EventRouter` - `org.apache.kafka.connect.transforms.RegexRouter` - `org.apache.kafka.connect.transforms.TimestampRouter` - `io.confluent.connect.transforms.MessageTimestampRouter` - `io.confluent.connect.transforms.ExtractTopic$Key` - `io.confluent.connect.transforms.ExtractTopic$Value` #### NOTE When the connector is not configured with `upsertEnabled` or `deleteEnabled`, these SMTs can be used without any issue. ## Supported data types The following list contains the supported BigQuery data types and the associated connector mapping. Note that this mapping applies when the connector creates a table automatically or updates schemas (that is, if either `auto.create.tables` or `auto.update.schemas` is set to `true`). | BigQuery Data Type | Connector Mapping | |----------------------|-------------------------| | STRING | String | | INTEGER | INT8 | | INTEGER | INT16 | | INTEGER | INT32 | | INTEGER | INT64 | | FLOAT | FLOAT32 | | FLOAT | FLOAT64 | | BOOLEAN | Boolean | | BYTES | Bytes | | TIMESTAMP | Logical TIMESTAMP | | TIME | Logical TIME | | DATE | Logical DATE | | FLOAT | Logical Decimal | | DATE | Debezium Date | | TIME | Debezium MicroTime | | TIME | Debezium Time | | TIMESTAMP | Debezium MicroTimestamp | | TIMESTAMP | Debezium TIMESTAMP | | TIMESTAMP | Debezium ZonedTimestamp | ## Install the BigQuery connector You can install this connector by using the [confluent connect plugin install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html) command, or by manually downloading the ZIP file. ### Prerequisites - You must install the connector on every machine where Connect will run. - If you want to install the connector using Confluent Marketplace, you must install the [Confluent Hub Client](https://docs.confluent.io/home/connect/confluent-hub/client.html). This is installed by default with Confluent Enterprise. - Kafka Broker: Confluent Platform 3.3.0 or later, or Kafka 0.11.0 or later - Connect: Confluent Platform 3.3.0 or later, or Kafka 0.11.0 or later - Java 1.8 - Active Google Cloud Platform (GCP) account with authorization to create resources #### IMPORTANT Streaming into BigQuery is not available with the Google Cloud free tier. If you try to use streaming without enabling billing, you receive the following error: `BigQuery: Streaming insert is not allowed in the free tier.`. For more details, see [Streaming data into BigQuery](https://cloud.google.com/bigquery/streaming-data-into-bigquery). ### Install the connector using the Confluent CLI To install the `latest` connector version using [Confluent Hub Client](https://www.confluent.io/hub/), navigate to your Confluent Platform installation directory and run the following command: ```bash confluent connect plugin install wepay/kafka-connect-bigquery:latest ``` You can install a specific version by replacing `latest` with a version number as shown in the following example: ```bash confluent connect plugin install wepay/kafka-connect-bigquery:2.6.0 ``` ### Install the connector manually [Download and extract the ZIP file](https://www.confluent.io/hub/wepay/kafka-connect-bigquery) for your connector and then follow the manual connector installation [instructions](/kafka-connectors/self-managed/userguide.html#connect-installing-plugins). ## License The Google Cloud BigQuery Sink Connector is an open source connector and does not require a Confluent Enterprise License. ## Linux on IBM Z (s390x) support Starting with Confluent Platform 8.2, this connector supports Linux on IBM Z (s390x). The connector supports the same capability available on x86_64 unless otherwise noted. For more information, see [Linux on IBM Z (s390x) support](https://docs.confluent.io/platform/current/installation/versions-interoperability.html#linux-on-ibm-z-s390x-support). ## Configuration properties For a complete list of configuration properties for this connector, see [Configuration Reference for Google BigQuery Sink Connector for Confluent Platform (Legacy)](kafka_connect_bigquery_config.md#kafka-connect-bigquery-config). ## Upgrading to 2.x.x The following changes aren’t backward compatible in the BigQuery connector: - `datasets` was removed and `defaultDataset` has been introduced. The connector now infers the dataset from the topic name if the topic is in the form `:`. If the topic name is in the form ``, the connector defaults to `defaultDataset`. - `topicsToTables` was removed. You should use SMT [RegexRouter](https://docs.confluent.io/platform/current/connect/transforms/regexrouter.html) to route topics to tables. - `autoUpdateSchemas` was replaced by `allowNewBigQueryFields` and `allowBigQueryRequiredFieldRelaxation`. - `value.converter.enhanced.avro.schema.support` should be set to false or removed. If this property is not removed or set to false, you may receive the following error: ```text Invalid field name "com.examples.project-super-important.v1.MyData". Fields must contain only letters, numbers, and underscores, start with a letter or underscore, and be at most 300 characters long. ``` ## Quick start The Confluent BigQuery Sink connector can stream table records into BigQuery from Kafka topics. These records are streamed at high throughput rates to facilitate analytical queries in near real-time. For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see [Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster). ### Install the connector To install the BiqQuery connector, complete the following steps. Note that to run the following steps, you must have Confluent Platform running locally. 1. Navigate to your Confluent Platform installation directory and enter the following command: ```bash confluent connect plugin install wepay/kafka-connect-bigquery:latest ``` 2. Use the Confluent CLI to restart Connect as adding a new connector plugin requires restarting Kafka Connect. ```bash confluent local services connect stop && confluent local services connect start Using CONFLUENT_CURRENT: /Users/username/Sandbox/confluent-snapshots/var/confluent.NuZHxXfq Starting Zookeeper Zookeeper is [UP] Starting Kafka Kafka is [UP] Starting Schema Registry Schema Registry is [UP] Starting Kafka REST Kafka REST is [UP] Starting Connect Connect is [UP] ``` 1. Verify the BigQuery Sink Connector plugin has been installed correctly and recognized by the plugin loader: ```bash curl -sS localhost:8083/connector-plugins | jq .[].class | grep BigQuerySinkConnector "com.wepay.kafka.connect.bigqueryl.BigQuerySinkConnector" ``` ### Set up the Google Cloud BigQuery connector Use the topics in this section to set up the BigQuery Sink connector for Confluent Platform. #### Prerequisites The following prerequisites are required before setting up the BigQuery connector. - An active Google Cloud account with authorization to create resources. - A BigQuery project. You can create the project using the [Google Cloud Console](https://console.cloud.google.com/). - A [BigQuery dataset](https://cloud.google.com/bigquery/docs/datasets) in the project. - A service account that can access the BigQuery project containing the dataset. You can create this service account in the [Google Cloud Console](https://console.cloud.google.com/). - The service account must have access to the BigQuery project containing the dataset. You create and download a key when creating a service account. You must download the key as a JSON file as shown in the following example: ```json { "type": "service_account", "project_id": "confluent-842583", "private_key_id": "...omitted...", "private_key": "-----BEGIN PRIVATE ...omitted... =\n-----END PRIVATE KEY-----\n", "client_email": "confluent2@confluent-842583.iam.gserviceaccount.com", "client_id": "...omitted...", "auth_uri": "https://accounts.google.com/oauth2/auth", "token_uri": "https://oauth2.googleapis.com/token", "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/certs", "client_x509_cert_url": "https://www.googleapis.com/robot/metadata/confluent2%40confluent-842583.iam.gserviceaccount.com" } According to `GCP specifications `__, the service account will either need the **BigQueryEditor** primitive IAM role or the **bigquery.dataEditor** predefined IAM role. The minimum permissions are as follows: .. code-block:: text bigquery.datasets.get bigquery.tables.create bigquery.tables.get bigquery.tables.getData bigquery.tables.list bigquery.tables.update bigquery.tables.updateData ``` ### Start the BigQuery Sink connector To start the BigQuery Sink Connector, complete the following steps: 1. Create the file `register-kcbd-connect-bigquery.json` to store the connector configuration. **Connect Distributed REST quick start connector properties:** ```json { "name": "kcbq-connect1", "config": { "connector.class": "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector", "tasks.max" : "1", "topics" : "kcbq-quickstart1", "sanitizeTopics" : "true", "autoCreateTables" : "true", "allowNewBigQueryFields" : "true", "allowBigQueryRequiredFieldRelaxation" : "true", "schemaRetriever" : "com.wepay.kafka.connect.bigquery.retrieve.IdentitySchemaRetriever", "project" : "confluent-243016", "defaultDataset" : "ConfluentDataSet", "keyfile" : " /Users/titomccutcheon/dev/confluent_fork/kafka-connect-bigquery/kcbq-connector/quickstart/properties/confluent-243016-384a24e2de1a.json", "transforms" : "RegexTransformation", "transforms.RegexTransformation.type" : "org.apache.kafka.connect.transforms.RegexRouter", "transforms.RegexTransformation.regex" : "(kcbq_)(.*)", "transforms.RegexTransformation.replacement" : "$2" } } ``` Note that the `project` key is the `id` value of the BigQuery project : in Google Cloud. For `datasets`, the value `ConfluentDataSet` is the ID of the dataset entered by the user during Google Cloud dataset creation.\`\`keyfile\`\` is the service account key JSON file location.
If you don’t want this connector to create a BigQuery table automatically, create a BigQuery table with `Partitioning: Partition by ingestion time` and a proper schema.
Also, note that the properties prefixed with `transforms` are used to set up SMTs. The following is an example regex router SMT that strips `kcbq_` from the topic name. Replace with relevant regex to replace the topic of each sink record with destination dataset and table name in the format `:` or only the destination table name in the format `` 1. Start the connector. ```text curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-kcbd-connect-bigquery.json ``` ### Start your Kafka producer To start your Kafka producer, complete the following steps: 1. Go to the Kafka bin folder and start a producer in a new terminal session. 2. Type the following command which waits on terminal input. ```text ./kafka-avro-console-producer --broker-list localhost:9092 --topic kcbq-quickstart1 --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}' ``` 3. Enter text for two test records, and press `Enter` after typing each line. ```text ./kafka-avro-console-producer --broker-list localhost:9092 --topic kcbq-quickstart1 --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}' {"f1":"Testing the Kafka-BigQuery Connector!"} {"f1":"Testing the Kafka-BigQuery Connector for a second time!"} ``` ### Checking results in BigQuery To the check the results in BigQuery, complete the following steps: 1. Go to the BigQuery editor in Google Cloud. 2. Enter the following SQL `SELECT` statement. ```text SELECT * FROM ConfluentDataSet.quickstart1; ``` #### NOTE `ConfluentDataSet` is the dataset ID and `quickstart1` is the name of the BigQuery table taken from the Kafka topic. In this case, the SMT strips `kcbq_` from the topic: the Connector converts this topic name `kcbq_quickstart1` to table name `quickstart1`. ### Clean up resources To clean up resources in BiqQuery, complete the following steps: 1. Delete the connector: ```bash curl -X DELETE localhost:8083/connectors/kcbq-connect1 ``` 2. Stop Confluent services: ```bash confluent local stop ``` ## Partition features To use timestamp partitioning by field name, you must set the `bigQueryPartitionDecorator` to `false`, and then set the `timestampPartitionFieldName` property to the field name that contains the partitioning timestamps (for example, `timestampPartitionFieldName=f2`). With the BigQuery console, you can use the following query to output a list of existing partitions: ```text SELECT f2 as pt, FORMAT_TIMESTAMP("%Y%m%d", f2) as partition_id FROM `PROJECT_ID.DATASET_ID.TABLE_ID` GROUP BY f2 ORDER BY f2 ``` The output lists all rows that have different dates and an additional `partition_id` column. Records that have a timestamp within a day’s range will have the same `partition_id`.