Skip to main content

​
Overview

The Google BigQuery Sink Connector writes Kafka Connect records from Kafka topics to Google BigQuery. By default, the topic name becomes the destination table and defaultDataset supplies the dataset. You can also define explicit topic-to-table mappings or rewrite a topic name as dataset:table to select the destination. The connector supports automatic table creation, schema updates, time partitioning, clustering, concurrent writes, and an optional GCS batch-load path.

​
Prerequisites

  • Enable the BigQuery API in the Google Cloud project and grant the connector service account the permissions required for the target project, datasets, tables, and any GCS bucket used for batch loading.

​
License

Licensed under the Apache License 2.0.

​
Quick Start

Prepare a Connect Cluster, Kafka, a Google Cloud project, and a BigQuery dataset. Confirm that the Connect Worker can reach Google Cloud. For connector creation, updates, and status checks, see AutoMQ Manage Connectors.
Replace the placeholders with actual resources. keySource=FILE means that keyfile is a service-account JSON file path. Use keySource=JSON when passing JSON content instead. Do not commit credentials or write them to logs.

​
Configuration

​
Kafka Connect Framework

​
connector.class

Specifies the BigQuery Sink Connector implementation class to load.
  • Type: class
  • Default: None
  • Importance: High
  • Valid values / notes: Required; use com.wepay.kafka.connect.bigquery.BigQuerySinkConnector.

​
Consumption

​
topics

Comma-separated list of topics to consume.
  • Type: list
  • Default: empty list
  • Importance: High
  • Valid values / notes: Mutually exclusive with topics.regex.

​
topics.regex

Java regular expression used to select topics.
  • Type: string
  • Default: empty string
  • Importance: High
  • Valid values / notes: Mutually exclusive with topics.

​
Destination and Authentication

​
project

Google Cloud project to write to.
  • Type: string
  • Default: None
  • Importance: High
  • Valid values / notes: Required.

​
defaultDataset

Default BigQuery dataset.
  • Type: string
  • Default: None
  • Importance: High
  • Valid values / notes: Required; dataset:table routing can override it.

​
keyfile

Service-account JSON file path or JSON content.
  • Type: password
  • Default: null
  • Importance: Medium
  • Valid values / notes: Set a file path for keySource=FILE or JSON content for keySource=JSON. Do not configure it when using application default credentials. Protect the credential.

​
keySource

Specifies whether keyfile is a file path or JSON content.
  • Type: string
  • Default: FILE
  • Importance: Medium
  • Valid values / notes: FILE, JSON, or APPLICATION_DEFAULT. With APPLICATION_DEFAULT, the connector obtains Google Cloud application default credentials from the Worker environment, and keyfile should not be set.

​
Tables and Schemas

​
autoCreateTables

Automatically create a destination table when it does not exist.
  • Type: boolean
  • Default: true
  • Importance: High
  • Valid values / notes: Requires schemaRetriever.

​
schemaRetriever

SchemaRetriever implementation used to create tables or update schemas.
  • Type: class
  • Default: com.wepay.kafka.connect.bigquery.retrieve.IdentitySchemaRetriever
  • Importance: Medium
  • Valid values / notes: The implementation must have a no-argument constructor. A custom implementation can replace the default.

​
allowNewBigQueryFields

Allow new fields to be added during later schema updates.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / notes: Requires schemaRetriever.

​
allowBigQueryRequiredFieldRelaxation

Allow a BigQuery REQUIRED field to become NULLABLE.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / notes: Requires schemaRetriever.

​
allowSchemaUnionization

Union the existing table schema with record schemas in the current batch during schema updates.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / notes: Requires schemaRetriever. Validate upstream schema quality before enabling it; BigQuery does not support dropping columns or arbitrarily changing existing column types.

​
Updates and Deletes

​
upsertEnabled

Use Kafka record keys, intermediate tables, and periodic MERGE operations to update destination rows.
  • Type: boolean
  • Default: false
  • Importance: Low
  • Valid values / notes: Requires kafkaKeyFieldName and at least one enabled MERGE trigger.

​
deleteEnabled

Use Kafka record keys to merge tombstone records into destination-table delete operations.
  • Type: boolean
  • Default: false
  • Importance: Low
  • Valid values / notes: Requires kafkaKeyFieldName and at least one enabled MERGE trigger.

​
intermediateTableSuffix

Suffix added to intermediate table names used by upsert and delete processing.
  • Type: string
  • Default: tmp
  • Importance: Low
  • Valid values / notes: Must be non-empty. Intermediate table names begin with the destination table name and this suffix.

​
mergeIntervalMs

Time interval for triggering MERGE operations when upsert or delete is enabled, in milliseconds.
  • Type: long
  • Default: 60000
  • Importance: Low
  • Valid values / notes: Use a positive integer, or -1 to disable time-based triggering. mergeIntervalMs and mergeRecordsThreshold cannot both be -1.

​
mergeRecordsThreshold

Trigger a MERGE after the intermediate table accumulates this number of records.
  • Type: long
  • Default: -1
  • Importance: Low
  • Valid values / notes: Use a positive integer, or -1 to disable record-count triggering. mergeIntervalMs and mergeRecordsThreshold cannot both be -1.

​
Names, Fields, and Partitioning

​
sanitizeTopics

Sanitize topic-derived table names.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / notes: The resulting table name can differ from the topic name.

​
topic2TableMap

Explicitly map Kafka topics to BigQuery tables.
  • Type: string
  • Default: empty string
  • Importance: Low
  • Valid values / notes: Use topic-1:table-1,topic-2:table-2. This setting overrides sanitizeTopics; an unmapped topic still uses its topic name as the table name. Do not combine it with a regex SMT that rewrites topic names.

​
sanitizeFieldNames

Replace invalid field-name characters with underscores and prefix names beginning with a digit with an underscore.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / notes: Different source names can collide after sanitization.

​
kafkaKeyFieldName

Field name used to store the Kafka record key.
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / notes: No Kafka key field is written when empty.

​
kafkaDataFieldName

Field name used to store Kafka data and metadata.
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / notes: No metadata structure is added when empty.

​
bigQueryPartitionDecorator

Use a BigQuery partition decorator for date-partitioned writes.
  • Type: boolean
  • Default: true
  • Importance: High
  • Valid values / notes: Cannot be used with timestampPartitionFieldName.

​
bigQueryMessageTimePartitioning

Use the Kafka message timestamp instead of connector processing time for the partition date.
  • Type: boolean
  • Default: false
  • Importance: High
  • Valid values / notes: Records must contain a valid timestamp.

​
timestampPartitionFieldName

Field in the value used for timestamp partitioning.
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / notes: Cannot be used with bigQueryPartitionDecorator=true.

​
clusteringPartitionFieldNames

Comma-separated fields used for BigQuery clustering.
  • Type: list
  • Default: null
  • Importance: Low
  • Valid values / notes: Up to four fields; requires a partitioned table.

​
timePartitioningType

Time-partitioning granularity used when the connector creates tables.
  • Type: string
  • Default: DAY
  • Importance: Low
  • Valid values / notes: HOUR, DAY, MONTH, YEAR, or NONE. Existing tables are not altered to use this partitioning type.

​
partitionExpirationMs

Partition expiration applied when the connector creates tables, in milliseconds.
  • Type: long
  • Default: null
  • Importance: Low
  • Valid values / notes: Applies only to newly created tables and does not alter existing tables. Data in expired partitions is permanently deleted.

​
Throughput, Retries, and Batch Loading

​
threadPoolSize

Maximum number of concurrent BigQuery write threads per task.
  • Type: int
  • Default: 10
  • Importance: Medium
  • Valid values / notes: At least 1.

​
queueSize

Soft write-queue limit after which partitions are paused.
  • Type: long
  • Default: -1
  • Importance: High
  • Valid values / notes: -1 means unlimited; partitions resume after the queue drains.

​
bigQueryRetry

Number of retries per request for BigQuery backend or quota errors.
  • Type: int
  • Default: 0
  • Importance: Medium
  • Valid values / notes: At least 0.

​
bigQueryRetryWait

Minimum wait between retries, in milliseconds.
  • Type: long
  • Default: 1000
  • Importance: Medium
  • Valid values / notes: At least 0.

​
max.retries

Maximum number of times the task reprocesses the current record batch after a retriable write failure.
  • Type: int
  • Default: 10
  • Importance: Medium
  • Valid values / notes: At least 1. This controls task-level batch retries and is separate from bigQueryRetry for an individual BigQuery request.

​
enableBatchLoad

List of topics written through the GCS batch-load path.
  • Type: list
  • Default: empty list
  • Importance: Low
  • Valid values / notes: Beta feature; requires gcsBucketName.

​
gcsBucketName

GCS bucket for batch-load objects.
  • Type: string
  • Default: empty string
  • Importance: High
  • Valid values / notes: Required when batch loading is enabled.

​
gcsFolderName

Folder prefix for GCS batch-load objects.
  • Type: string
  • Default: empty string
  • Importance: Medium
  • Valid values / notes: Used only when batch loading is enabled.

​
batchLoadIntervalSec

Interval between GCS-to-BigQuery load attempts, in seconds.
  • Type: int
  • Default: 120
  • Importance: Low
  • Valid values / notes: Used only when batch loading is enabled.

​
autoCreateBucket

Automatically create the GCS bucket when batch loading starts and the bucket does not exist.
  • Type: boolean
  • Default: true
  • Importance: Medium
  • Valid values / notes: Used only when batch loading is enabled.

​
convertDoubleSpecialValues

Convert special floating-point values to finite values accepted by BigQuery.
  • Type: boolean
  • Default: false
  • Importance: Low
  • Valid values / notes: Confirm how downstream consumers should interpret the converted values.

​
avroDataCacheSize

Avro schema conversion cache size.
  • Type: int
  • Default: 100
  • Importance: Low
  • Valid values / notes: At least 0.

​
allBQFieldsNullable

Translate produced BigQuery fields as NULLABLE; array fields remain REPEATED.
  • Type: boolean
  • Default: false
  • Importance: Low
  • Valid values / notes: Must be combined with required-field relaxation.

​
Best Practices

​
Control Concurrency and Backpressure for High-Throughput Writes

Applicable business scenario: The input rate can exceed BigQuery processing capacity, so queued work must be bounded and transient backend errors retried. Configuration example:
Key points: threadPoolSize controls concurrent writes. Partitions are paused after the queueSize soft limit is reached. Tune concurrency and retries against BigQuery quotas and acceptable latency.

​
Partition Tables by Event Time

Applicable business scenario: Records contain a business event timestamp and should be partitioned by that value rather than connector receive time. Configuration example:
Key points: Field-based partitioning is mutually exclusive with the default bigQueryPartitionDecorator=true. Use no more than four clustering fields on a partitioned table.

​
Monitoring

​
What to Monitor

Monitor Worker, Connector, and Task state; throughput, latency, offset commits, errors, retries, and failed tasks; and Worker JVM heap, GC, thread count, and CPU. When queueSize is enabled, watch backlog and partition pause/resume activity. Monitor DLQ activity only when error handling and a DLQ are deployed.

​
Import the Grafana Dashboard

Download the AutoMQ Connect Cluster Dashboard, configure a Prometheus data source with Kafka Connect metrics and matching labels, and import the JSON dashboard into Grafana.

​
Limitations

  • topics and topics.regex are mutually exclusive, and one must be configured.
  • timestampPartitionFieldName cannot be used with bigQueryPartitionDecorator=true.
  • clusteringPartitionFieldNames supports at most four fields and requires a partitioned table.
  • Field-name sanitization can map different source fields to the same BigQuery field name.
  • The connector does not claim exactly-once or duplicate-free BigQuery delivery.

​
FAQ

​
The connector reports a topic-selection conflict at startup. What should I do?

Keep either topics or topics.regex, and check whether a configuration provider, environment variable, or deployment layer injects the other property.

​
Startup reports that defaultDataset or project is missing. What should I do?

Neither property has a default. Set the Google Cloud project and BigQuery dataset, and verify service-account access.

​
Writes fail because the destination table does not exist. What should I do?

Keep autoCreateTables=true and configure a SchemaRetriever implementation, or create the table in advance when automatic creation is disabled.

​
Why are records routed to different datasets or tables?

Check whether an SMT rewrites the topic to dataset:table and whether sanitizeTopics is enabled. Without routing, the connector uses defaultDataset and the topic name.

​
What should I check when write latency keeps increasing?

Check BigQuery quotas, errors, and retry counts, then inspect threadPoolSize, queueSize, and Task state. If the queue does not drain, reduce input rate or adjust concurrency and retries carefully.