Google BigQuery Sink Connector for Confluent Platform¶
Note
If you’re using Confluent Cloud, see Google BigQuery Sink Connector for Confluent Cloud for the Confluent Cloud Quick Start.
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.
Important
Google BigQuery Sink Connector version 2.0.0 is not backward compatible with 1.x.x versions. See the Upgrading to 2.0.0 section for more information.
Features¶
- At least once delivery
- Dead Letter Queue
- Multiple tasks
- Insert operations and duplicate detection
- Insertion of records one at a time
- Streaming from a list of topics
- Internal thread pool
Note
If you’re planning to use multiple connectors with a high number of tasks, be sure to review BigQuery rate limits.
At least once delivery¶
This connector guarantees that records are delivered at least once from the Kafka topic.
Dead Letter Queue¶
This connector supports the Dead Letter Queue (DLQ) functionality. For information about accessing and using the DLQ, see Confluent Platform Dead Letter Queue.
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 connector supports insert operations and tries to detect duplicates. See BigQuery troubleshooting for more information.
Insertion of records one at a time¶
The connector uses the BigQuery insertAll streaming api which inserts records one at a time. The records 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
The internal thread pool defaults to 10 threads, which is configurable.
Limitations¶
The BigQuery Sink connector has the following limitations:
- The connector doesn’t support schemas with recursion.
When the connector is configured with
upsertEnabled
ordeleteEnabled
, it does not support Single Message Transformations (SMTs) 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
ordeleteEnabled
, these SMTs can be used without any issue.
Install the BigQuery Connector¶
You can install this connector by using the Confluent Hub client installation instructions 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 Hub, you must install the Confluent Hub Client. 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 data into BigQuery is not available using the GCP free tier. If you try
to use streaming without enabling billing, you will receive the following error:
BigQuery: Streaming insert is not allowed in the free tier.
See
Streaming data into BigQuery for more
details.
Install the connector using Confluent Hub¶
To install the latest
connector version using Confluent Hub Client, navigate to your Confluent Platform installation directory
and run the following command:
confluent-hub 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:
confluent-hub install wepay/kafka-connect-bigquery:2.5.0
Install the connector manually¶
Download and extract the ZIP file for your connector and then follow the manual connector installation instructions.
License¶
The GCP BigQuery Sink Connector is an open source connector and does not require a Confluent Enterprise License.
Configuration Properties¶
For a complete list of configuration properties for this connector, see Google BigQuery Sink Connector Configuration Properties.
Upgrading to 2.0.0¶
The following changes aren’t backward compatible in the BigQuery connector:
datasets
was removed anddefaultDataset
has been introduced. The connector now infers the dataset from the topic name if the topic is in the form<dataset>:<tableName>
. If the topic name is in the form<tablename>
, the connector defaults todefaultDataset
.topicsToTables
was removed. You should use SMT RegexRouter to route topics to tables.autoUpdateSchemas
was replaced byallowNewBigQueryFields
andallowBigQueryRequiredFieldRelaxation
.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: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.
Note
For an example of how to get Kafka Connect connected to Confluent Cloud, see Distributed Cluster.
Install the Connector¶
To install the BiqQuery connector, complete the following steps:
Note
For the following tutorial, you must have Confluent Platform running locally.
Navigate to your Confluent Platform installation directory and enter the following command:
confluent-hub install wepay/kafka-connect-bigquery:latest
Use the Confluent CLI to restart Connect as adding a new connector plugin requires restarting Kafka Connect.
Tip
The command syntax for the Confluent CLI development commands changed in 5.3.0.
These commands have been moved to confluent local
. For example, the syntax for confluent start
is now
confluent local services start
. For more information, see confluent local.
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]
Verify the BigQuery Sink Connector plugin has been installed correctly and recognized by the plugin loader:
curl -sS localhost:8083/connector-plugins | jq .[].class | grep BigQuerySinkConnector "com.wepay.kafka.connect.bigqueryl.BigQuerySinkConnector"
Setting up the GCP BigQuery Connector¶
Important
Streaming into BigQuery not available via the GCP free tier. If you attempt
to use streaming without enabling billing, you receive the following error:
BigQuery: Streaming insert is not allowed in the free tier.
See
Streaming data into BigQuery for
details.
Prerequisites¶
The following prerequisites are required before setting up the BigQuery connector.
An active GCP account with authorization to create resources.
A BigQuery project. You can create the project using the Google Cloud Console.
A BigQuery dataset 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.
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:
{ "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 <https://cloud.google.com/bigquery/docs/access-control>`__, 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
Starting the BigQuery Sink Connector¶
To start the BigQuery Sink Connector, complete the following steps:
Create the file
register-kcbd-connect-bigquery.json
to store the connector configuration.Connect Distributed REST quick start connector properties:
{ "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
The
project
key is theid
value of the BigQuery project in GCP. Fordatasets
, the valueConfluentDataSet
is the ID of the dataset entered by the user during GCP 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.Note
The properties prefixed with
transforms
are used to set up SMTs. The following is an example regex router SMT that stripskcbq_
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<dataset>:<tableName>
or only the destination table name in the format<tableName>
Start the connector.
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-kcbd-connect-bigquery.json
Starting your Kafka producer¶
To start your Kafka producer, complete the following steps:
Go to the Kafka bin folder and start a producer in a new terminal session.
Type the following command which waits on terminal input.
./kafka-avro-console-producer --broker-list localhost:9092 --topic kcbq-quickstart1 --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}'
Enter text for two test records, and press
Enter
after typing each line../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:
Go to the BigQuery editor in GCP.
Enter the following SQL
SELECT
statement.SELECT * FROM ConfluentDataSet.quickstart1;
Note
ConfluentDataSet
is the dataset ID andquickstart1
is the name of the BigQuery table taken from the Kafka topic. In this case, the SMT stripskcbq_
from the topic: the Connector converts this topic namekcbq_quickstart1
to table namequickstart1
.
Cleaning up resources¶
To clean up resources in BiqQuery, complete the following steps:
Delete the connector:
curl -X DELETE localhost:8083/connectors/kcbq-connect1
Stop Confluent services:
confluent local stop
Partitioning 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:
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
.