• Debezium Bigquery Consumers
  • bigquerybatch Consumer
  • bigquerystream Consumer
• Install from source

This project adds Bigquery consumers to Debezium Server. This consumer replicates RDBMS CDC events to Bigquery in real time.

Writes debezium events to Bigquery using BigQuery Storage Write API. It groups CDC events and appends to destination BigQuery table using BigQuery Write API .

Streams debezium events to Bigquery using the Storage Write API.

For detailed information on data type mapping between Debezium and BigQuery, refer to the documentation

By default, Bigquery Streaming consumer is running with append mode debezium.sink.iceberg.upsert=false. Upsert mode uses source Primary Key and does upsert on target table(delete followed by insert). For the tables without Primary Key consumer falls back to append mode.

With upsert mode data deduplication is done. Deduplication is done based on __source_ts_ms value and event type __op . its is possible to change this field using debezium.sink.bigquerystream.upsert-dedup-column=__source_ts_ms (Currently only Long field type supported.)

Operation type priorities are {"c":1, "r":2, "u":3, "d":4}. When two records with same key and same __source_ts_ms values received then the record with higher __op priority is kept and added to destination table and duplicate record is dropped from stream.

debezium.format.value=json
debezium.format.key=json
debezium.format.schemas.enable=true

Bigquery consumers requires event flattening, please see debezium feature

debezium.transforms=unwrap
debezium.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
debezium.transforms.unwrap.add.fields=op,table,lsn,source.ts_ms
debezium.transforms.unwrap.add.headers=db
debezium.transforms.unwrap.delete.handling.mode=rewrite

Debezium extracts database events in real time and this could cause too frequent commits or too many small files which is not optimal for batch processing especially when near realtime data feed is sufficient. To avoid this problem following batch-size-wait classes are used.

Batch size wait adds delay between consumer calls to increase total number of events received per call and meanwhile events are collected in memory. This setting should be configured together with debezium.source.max.queue.size and debezium.source.max.batch.size debezium properties

This is default configuration, by default consumer will not use any wait. All the events are consumed immediately.

MaxBatchSizeWait uses debezium metrics to optimize batch size. MaxBatchSizeWait periodically reads streaming queue current size and waits until number of events reaches to max.batch.size or until debezium.sink.batch.batch-size-wait.max-wait-ms.

Maximum wait and check intervals are controlled by debezium.sink.batch.batch-size-wait.max-wait-ms , debezium.sink.batch.batch-size-wait.wait-interval-ms properties.

example setup to receive ~2048 events per commit. maximum wait is set to 30 seconds, streaming queue current size checked every 5 seconds

debezium.sink.batch.batch-size-wait=MaxBatchSizeWait
debezium.sink.batch.metrics.snapshot-mbean=debezium.postgres:type=connector-metrics,context=snapshot,server=testc
debezium.sink.batch.metrics.streaming-mbean=debezium.postgres:type=connector-metrics,context=streaming,server=testc
debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
debezium.source.max.batch.size=2048;
debezium.source.max.queue.size=16000";
debezium.sink.batch.batch-size-wait.max-wait-ms=30000
debezium.sink.batch.batch-size-wait.wait-interval-ms=5000

This implementation saves CDC offset to a bigquery table.

debezium.source.offset.storage=io.debezium.server.bigquery.offset.BigqueryOffsetBackingStore
debezium.source.offset.storage.bigquery.table-name=debezium_offset_storage_custom_table

This implementation saves database history to a bigquery table.

debezium.source.database.history=io.debezium.server.bigquery.history.BigquerySchemaHistory
debezium.source.database.history.bigquery.table-name=__debezium_database_history_storage_test_table

quarkus.log.level=INFO
# Ignore messages below warning level from Jetty, because it's a bit verbose
quarkus.log.category."org.eclipse.jetty".level=WARN
#

• Requirements:
  • JDK 11
  • Maven
• Clone from repo: git clone https://github.com/memiiso/debezium-server-bigquery.git
• From the root of the project:
  • Build and package debezium server: mvn -Passembly -Dmaven.test.skip package
  • After building, unzip your server distribution: unzip debezium-server-bigquery-dist/target/debezium-server-bigquery-dist*.zip -d appdist
  • cd into unzipped folder: cd appdist
  • Create application.properties file and config it: nano conf/application.properties, you can check the example configuration in application.properties.example
  • Run the server using provided script: bash run.sh

The Memiiso community welcomes anyone that wants to help out in any way, whether that includes reporting problems, helping with documentation, or contributing code changes to fix bugs, add tests, or implement new features.