> ## Documentation Index
> Fetch the complete documentation index at: https://docs.automq.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Google BigQuery Sink Connector

> Configure and operate the Google BigQuery Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting.

## 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](../manage-connectors).

```properties theme={null}
connector.class=com.wepay.kafka.connect.bigquery.BigQuerySinkConnector
topics=orders
project=<gcp-project>
defaultDataset=<dataset>
keyfile=<service-account-json-path>
keySource=FILE
autoCreateTables=true
schemaRetriever=com.wepay.kafka.connect.bigquery.retrieve.IdentitySchemaRetriever
```

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

```properties theme={null}
connector.class=com.wepay.kafka.connect.bigquery.BigQuerySinkConnector
topics=orders
project=<gcp-project>
defaultDataset=<dataset>
keyfile=<service-account-json-path>
autoCreateTables=false
threadPoolSize=10
queueSize=1000
bigQueryRetry=3
bigQueryRetryWait=1000
```

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

```properties theme={null}
connector.class=com.wepay.kafka.connect.bigquery.BigQuerySinkConnector
topics=orders
project=<gcp-project>
defaultDataset=<dataset>
keyfile=<service-account-json-path>
autoCreateTables=true
schemaRetriever=com.wepay.kafka.connect.bigquery.retrieve.IdentitySchemaRetriever
bigQueryPartitionDecorator=false
timestampPartitionFieldName=event_time
clusteringPartitionFieldNames=customer_id,region
```

**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](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), 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.