BigQuery examples#

The BigQuery quickstart owns the basic DataStream read and write jobs. The examples below keep only the parts that change for a particular use case.

DataStream source#

Start with the Quickstart’s bounded table read. The cases below change its projection, billing project, snapshot, query, or stream assignment.

Reading one column of a large table#

The two push-down knobs are applied by BigQuery when the read session is created, so what they exclude never leaves it — and the columns you leave out are not scanned, which is what the read is charged for.

Schema readerSchema =
        new Schema.Parser()
                .parse(
                        "{\"type\":\"record\",\"name\":\"Event\",\"fields\":["
                                + "{\"name\":\"user_id\",\"type\":\"long\"}]}");

Source<GenericRecord, ?, ?> source =
        BigQuerySource.<GenericRecord>builder()
                .table(TableDestination.of("my-project", "analytics", "events"))
                .deserializer(BigQueryRowDeserializationSchema.genericRecord(readerSchema))
                .selectedFields("user_id")
                .rowRestriction("event_date = '2026-08-01' AND country = 'JP'")
                .build();

The reader schema names only the column being read. A row’s other columns are dropped by Avro’s schema resolution before the record is built — and here they never left BigQuery in the first place.

Reading a public dataset#

A read session belongs to a project, and that is the project it is billed to. Reading a table you do not own — a public dataset, or another team’s — means naming your own project as the payer:

BigQuerySource.<GenericRecord>builder()
        .table(TableDestination.of("bigquery-public-data", "samples", "shakespeare"))
        .parentProject("my-project")
        .deserializer(BigQueryRowDeserializationSchema.genericRecord(readerSchema))
        .build();

Without parentProject the session would be created in bigquery-public-data, where you have no permission to create one.

Reading a table as it was#

snapshotTime reads the table as of an instant, from BigQuery’s time-travel window. Two jobs given the same instant read the same rows, whatever has been written since — which is what makes a re-run reproducible rather than merely repeated.

BigQuerySource.<GenericRecord>builder()
        .table(TableDestination.of("my-project", "my_dataset", "accounts"))
        .snapshotTime(Instant.parse("2026-08-01T00:00:00Z"))
        .deserializer(BigQueryRowDeserializationSchema.genericRecord(readerSchema))
        .build();

Note that a read session pins its own snapshot at creation regardless, so a job that does not set this still reads one consistent view of the table — just whichever one existed when it started.

Reading a view#

A view cannot be read as a table — the Storage Read API reads storage, and a view has none. Run it as a query instead, and the source reads the table its result lands in.

BigQuerySource.<GenericRecord>builder()
        .query("SELECT id, name FROM `my-project.my_dataset.active_accounts`")
        .parentProject("my-project")
        .deserializer(BigQueryRowDeserializationSchema.genericRecord(readerSchema))
        .build();

parentProject is required here rather than optional: no table is named, so nothing else says which project runs the query job and is billed for it. By default the result goes to BigQuery’s own anonymous dataset, which expires it in about a day and charges no storage for it — nothing to create and nothing to clean up.

Prune inside the query rather than with selectedFields: those are applied to the result, so they cannot make the query itself cheaper, and a query source pays for both scans. The trade-offs and the constraints of each landing place are under Reading a query or a view.

Reading a view without writing the query#

If a job is pointed at names it does not control — a catalog where some are tables and some are views — materializeViews() handles both without the job having to know which is which.

BigQuerySource.<GenericRecord>builder()
        .table(TableDestination.of("my-project", "my_dataset", "active_accounts"))
        .materializeViews()
        .selectedFields("id", "region")
        .deserializer(BigQueryRowDeserializationSchema.genericRecord(readerSchema))
        .build();

A view is materialized and read; an ordinary table is read directly, with nothing billed for a query. It is off by default because it costs one metadata call per job to tell the two apart, and because materializing bills a query nobody wrote. selectedFields is folded into the generated SELECT, so a view is not scanned for columns that would only be discarded.

Landing a query result in your own dataset#

Name a dataset when the anonymous one will not do — because something outside the job has to read the result, or because a cached results table is not a dependency you want to take.

BigQuerySource.<GenericRecord>builder()
        .query(
                "SELECT o.id, c.region FROM `my-project.sales.orders` o"
                        + " JOIN `my-project.sales.customers` c USING (customer_id)")
        .parentProject("my-project")
        .queryResultDataset("scratch")
        .deserializer(BigQueryRowDeserializationSchema.genericRecord(readerSchema))
        .build();

The dataset must already exist and be in the query’s own location. The connector creates a table there with a one-day expiration and does not delete it earlier: teardown also runs on a JobManager failover, where the restored job is still reading the read session that table backs.

Asking for more read streams#

A read stream is read by one subtask at a time, and a subtask takes the next stream as soon as it finishes one. Over-provisioning is therefore how the work spreads evenly: with as many streams as subtasks, one slow stream leaves a subtask idle at the end.

BigQuerySource.<GenericRecord>builder()
        .table(TableDestination.of("my-project", "my_dataset", "events"))
        .deserializer(BigQueryRowDeserializationSchema.genericRecord(readerSchema))
        .preferredMinStreamCount(3 * env.getParallelism())
        .build();

BigQuery decides the actual count and may give fewer — a small table is read by one stream however many are asked for. The measured behaviour of both knobs is under Assignment and stream count.

DataStream sink#

Start with the Quickstart’s default-stream write. The cases below change destination routing, delivery method, or table creation.

A table per day#

The writer context carries the record’s event timestamp, which makes time-based routing expressible without the record carrying the routing key. The dynamic destinations guide defines the shared resolver contract and compares its resource lifetime with the other sinks. This resolver caches one destination per UTC day and falls back to the record’s own timestamp when the writer context has none.

public class DailyTableResolver implements DestinationResolver<OrderEvent> {

    private static final DateTimeFormatter SUFFIX = DateTimeFormatter.ofPattern("yyyyMMdd");

    private final String project;
    private final String dataset;
    private final String prefix;

    // One entry per day rather than one TableDestination per record. A plain HashMap is enough:
    // the writer is single-threaded per subtask, and this resolver is never shared across them.
    private final Map<LocalDate, TableDestination> cache = new HashMap<>();

    public DailyTableResolver(String project, String dataset, String prefix) {
        this.project = project;
        this.dataset = dataset;
        this.prefix = prefix;
    }

    @Override
    public TableDestination resolve(OrderEvent element, SinkWriter.Context context) {
        Long eventTime = context.timestamp();
        // Null when nothing assigned the record a timestamp — a processing-time job, or a
        // source
        // with no timestamp assigner. Falling back to the record's own field keeps such a
        // record
        // routed rather than unroutable.
        Instant instant =
                eventTime != null ? Instant.ofEpochMilli(eventTime) : element.createdAt();
        LocalDate day = instant.atZone(ZoneOffset.UTC).toLocalDate();
        return cache.computeIfAbsent(
                day,
                d -> TableDestination.of(project, dataset, prefix + "_" + d.format(SUFFIX)));
    }
}

BigQuerySink.<OrderEvent>builder()
        .destinationResolver(new DailyTableResolver("my-project", "my_dataset", "orders"))
        .serializer(serializer)
        .build();

Two things need planning when a resolver keeps producing new destinations. The default-stream and buffered-stream methods hold one writer per active destination, so DefaultStreamOptions and BufferedStreamOptions expose destinationIdleTimeout (one hour by default) to bound that local state. FILE_LOADS bounds each writer subtask to maxOpenDestinations active files (16 by default), finishes the least recently used file when that capacity is reached, and also finishes a file after destinationIdleTimeout (one minute by default). It retains at most maxPendingFiles finished and open files for the next commit (10,000 by default), failing before another file is opened if churn reaches that bound. A checkpoint finishes every remaining file and releases its conversion state. Every new table is also created on its first record under the default create disposition, so table auto-creation applies to every day this produces, not only the first.

Exactly-once#

Two of BigQuery’s three write methods are exactly-once, and they trade against each other rather than one being better. (Pub/Sub and Cloud Tasks are at-least-once with no exactly-once path — those services have no transactional publish.)

Both need streaming checkpointing in CheckpointingMode.EXACTLY_ONCE, which is Flink’s default and so needs no line in either job below — but a cluster setting execution.checkpointing.mode to AT_LEAST_ONCE has the job rejected when the graph is built, rather than silently downgraded.

Buffered streams#

Rows are appended to one Storage Write API buffered stream per (subtask, destination) at explicit offsets, invisible until a completed checkpoint makes exactly that checkpoint’s rows visible.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// The mode must be explicit: AUTOMATIC is rejected at graph construction, because resolving
// to
// streaming without checkpointing would leave buffered rows invisible forever.
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
env.enableCheckpointing(60_000);

env.fromSource(source, WatermarkStrategy.noWatermarks(), "orders")
        .sinkTo(
                BigQuerySink.<OrderEvent>builder()
                        .writeMethod(WriteMethod.STORAGE_API_EXACTLY_ONCE)
                        .destinationResolver(
                                new DailyTableResolver(
                                        "my-project", "my_dataset", "orders"))
                        .serializer(serializer)
                        .bufferedStreamOptions(BufferedStreamOptions.builder().build())
                        .build());

env.execute("bigquery-exactly-once");

bufferedStreamOptions(...) is required for this write method and rejected for the others, and every knob in it is defaulted — builder().build() is how to say “the defaults” out loud. The checkpoint interval is the visibility latency: rows land when the checkpoint that named them completes. Each active destination uses a dedicated connection and contributes its own stream creation and flush calls, so high-cardinality routing should use an idle timeout appropriate to its churn and must account for the Storage Write API’s stream-creation quota.

File loads#

Rows are staged as files on Cloud Storage — Avro by default — and loaded with BigQuery load jobs, which is free of streaming-insert cost and exactly-once in both execution modes.

env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
// The checkpoint is the load trigger, and each one modifies every active destination with a
// direct load or an overflow copy. 5 minutes is 288 modifications; below 2 minutes the sink
// rejects the job outright.
env.enableCheckpointing(300_000);

env.fromSource(source, WatermarkStrategy.noWatermarks(), "orders")
        .sinkTo(
                BigQuerySink.<OrderEvent>builder()
                        .writeMethod(WriteMethod.FILE_LOADS)
                        .destinationResolver(
                                new DailyTableResolver(
                                        "my-project", "my_dataset", "orders"))
                        .serializer(serializer)
                        .fileLoadsOptions(
                                FileLoadsOptions.builder()
                                        .stagingPath("gs://my-staging-bucket/flink-loads")
                                        .build())
                        .build());

env.execute("bigquery-file-loads");

Point stagingPath at a dedicated bucket, separate from checkpoint and savepoint storage, with a lifecycle rule deleting objects after a few days, so files orphaned by a hard failure expire on their own. Size the rule above the longest outage you intend to recover from: files a checkpoint still references are the data, and a streaming job restored after the rule expired them leaves its pending loads permanently failing.

Batch is the same builder with RuntimeExecutionMode.BATCH and no checkpointing — everything loads at end of input.

Redeploying an exactly-once job#

Never redeploy through discarded state. The two-phase commit puts rows and source positions in the same phase with no atomicity between them, so a writer restored with no state opens a loss window of at most one checkpoint: rows appended but not flushed, and committables checkpointed but not committed, stay invisible forever while the source may already have acked past them.

The sink cannot detect this — a writer restored with no state is indistinguishable from a new job — so the guard belongs in deployment tooling:

flink stop --savepointPath gs://my-savepoints <job-id>
flink run -s gs://my-savepoints/savepoint-xxxx my-job.jar

With the Flink Kubernetes Operator that is upgradeMode: savepoint (or last-state), never stateless. When state genuinely has to be dropped, rewind the source behind the last completed checkpoint so a potential loss becomes a duplicate instead, and make duplicates harmless downstream with an idempotent key plus MERGE or QUALIFY ROW_NUMBER().

The at-least-once write method has the opposite profile — it keeps the sink strictly ahead of the source, so discarding state can duplicate rows but cannot lose them. Neither method is uniformly safer; the BigQuery connector page sets their loss paths side by side.

Table auto-creation#

The default create disposition is CREATE_IF_NEEDED, so the first record for a missing table creates it from the serializer’s schema. tableCreateOptions(...) is what decides the rest of the table’s shape:

BigQuerySink.<OrderEvent>builder()
        .destinationResolver(new DailyTableResolver("my-project", "my_dataset", "orders"))
        .serializer(serializer)
        .tableCreateOptions(
                TableCreateOptions.builder()
                        .timePartitioning(TimePartitioningType.DAY, "created_at")
                        .timePartitioningExpiration(Duration.ofDays(90))
                        .clusteredFields(List.of("customer_id"))
                        .build())
        .build();

These apply at creation and never afterwards. An existing table is never modified by them, so adding partitioning to a running pipeline changes only the tables created from that point on. Use tableCreateOptionsProvider(...) instead when the settings vary per destination — it receives the TableDestination and returns the options for it.

Creation is idempotent across parallel subtasks (a 409 counts as success), so nothing needs coordinating — and a subtask the per-table quota rate-limits instead of answering 409 retries the creation within the recovery budget, so a wide parallelism costs a backoff rather than the job (see Losing the creation race). The credentials need bigquery.tables.create on the dataset; CREATE_NEVER turns a missing table into an immediate job failure instead, which is what to use when a missing table means a routing bug.

Creation is also the only moment a REQUIRED column can appear — BigQuery cannot add one to an existing table — so whichever column modes the serializer derives are decided here, durably.

Table source#

Reading a BigQuery table with SQL#

Create the physical table used by the quickstart’s read example and add one row:

bq query --use_legacy_sql=false \
  'CREATE TABLE IF NOT EXISTS `my-project.my_dataset.people` (id INT64, name STRING)'
bq query --use_legacy_sql=false \
  'INSERT `my-project.my_dataset.people` (id, name) VALUES (1001, "Ada")'

Register it as a bounded Flink table source:

CREATE TABLE people (
  id BIGINT,
  name STRING
) WITH (
  'connector' = 'bigquery',
  'project' = 'my-project',
  'dataset' = 'my_dataset',
  'table' = 'people'
);

SELECT name
FROM people;

The source finishes after reading one BigQuery snapshot, so the table works in a batch job or as the bounded side of a streaming job. Top-level projection is pushed into the Storage Read session, so the query above requests only name. Supported scalar comparisons and top-level scalar null predicates are also sent as Storage Read row restrictions. The connector keeps every pushed predicate as a Flink residual, so Flink rechecks every row BigQuery returns before producing the final result. Set scan.row-restriction in the DDL when a BigQuery-native expression outside that conservative subset is required.

Table sink#

Writing append-only rows with SQL#

The default Table sink uses storage-api-at-least-once and accepts an insert-only changelog. This complete example sets a checkpoint interval, declares the destination, and inserts one bounded row:

SET 'execution.checkpointing.interval' = '5 min';

CREATE TABLE analytics_events (
  event_id STRING,
  amount BIGINT
) WITH (
  'connector' = 'bigquery',
  'project' = 'my-project',
  'dataset' = 'analytics',
  'table' = 'events'
);

INSERT INTO analytics_events
SELECT event_id, amount
FROM (VALUES ('event-1', CAST(42 AS BIGINT))) AS staged_events(event_id, amount);

The DDL omits sink.write-method, so the default stream is selected. Keep the table schema and INSERT unchanged when another delivery method fits the job better.

Changing the write method#

For Storage Write API exactly-once, add this option to the same WITH clause:

'sink.write-method' = 'storage-api-exactly-once'

For FILE_LOADS, select the method and provide its one conditionally required option:

'sink.write-method' = 'file-loads',
'sink.file-loads.staging-path' = 'gs://my-staging-bucket/flink'

Both exactly-once methods need checkpointing in streaming execution, with execution.checkpointing.mode set to EXACTLY_ONCE and execution.checkpointing.checkpoints-after-tasks-finish.enabled set to true. Those two settings are Flink defaults, but a cluster override must be reverted for this job. Buffered streams make rows visible when a checkpoint commits them, while FILE_LOADS stages files in Cloud Storage and submits load jobs after each checkpoint. The Table connector delivery guarantees compare their recovery and visibility boundaries.

Change data capture#

CDC examples by source connector#

Each CDC source has its own ordering contract and example section:

SourceSequence profileExample status
Debezium PostgreSQLLast committed and current LSNAvailable below
Debezium MySQLSource UUID epoch, GTID transaction, binlog position, and rowAvailable below
TiCDC Debezium protocolTiDB commit TSO and cluster identityAvailable below
Debezium Spanner and native Spanner Change StreamsCommit timestamp, record sequence, and mod numberAvailable below

The implemented sections use complete source envelopes and do not substitute source timestamps or processing time when ordering metadata is absent.

Debezium MySQL CDC from Kafka#

Both examples consume Kafka values containing complete Debezium MySQL Avro envelopes and write the current row to a BigQuery table with a primary key. They forward connector, snapshot, gtid, pos, and row from the envelope’s source record.

MySQL Kafka source#

Use Flink’s Kafka connector and ConfluentRegistryAvroDeserializationSchema.forGeneric(...) to retain the complete envelope. Pass its Avro schema as debeziumEnvelopeSchema:

static KafkaSource<GenericRecord> kafkaSource(org.apache.avro.Schema debeziumEnvelopeSchema) {
    return KafkaSource.<GenericRecord>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("dbserver1.inventory.orders")
            .setGroupId("bigquery-cdc-mysql-orders")
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setValueOnlyDeserializer(
                    ConfluentRegistryAvroDeserializationSchema.forGeneric(
                            debeziumEnvelopeSchema, "http://schema-registry:8081"))
            .build();
}

A Debezium tombstone produces no element through Kafka’s value-only wrapper. Flink checkpoint restoration resumes from the Kafka positions stored in the checkpoint. Add a Kafka connector release compatible with the application’s Flink version and flink-avro-confluent-registry to the job artifact.

MySQL DataStream API#

Pass the KafkaSource<GenericRecord> above as kafkaSource and the Avro schema for one physical row as rowSchema:

env.enableCheckpointing(60_000);
env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Debezium MySQL orders")
        .sinkTo(
                BigQuerySink.<GenericRecord>builder()
                        .writeMethod(WriteMethod.STORAGE_API_AT_LEAST_ONCE)
                        .table(
                                TableDestination.of(
                                        "my-project", "analytics", "current_mysql_orders"))
                        .serializer(
                                new DebeziumEnvelopeSerializer(
                                        AvroRecordSerializationSchema.of(
                                                rowSchema,
                                                AvroSchemaOptions.builder()
                                                        .deriveRequiredColumns()
                                                        .build())))
                        .cdcTableOptions(
                                CdcTableOptions.builder()
                                        .primaryKeyColumns(Collections.singletonList("id"))
                                        .build())
                        .cdcTableReconciliationPolicy(
                                CdcTableReconciliationPolicy.RECONCILE)
                        .cdcOptions(
                                CdcOptions.<GenericRecord>builder(
                                                envelope ->
                                                        isDelete(envelope)
                                                                ? CdcChangeType.DELETE
                                                                : CdcChangeType.UPSERT)
                                        .sequenceNumberProvider(
                                                envelope ->
                                                        MYSQL_SEQUENCE_NUMBERS
                                                                .getSequenceNumber(
                                                                        sourceProperties(
                                                                                source(
                                                                                        envelope))))
                                        .build())
                        .build());

env.execute("debezium-mysql-to-bigquery-cdc");

The sequence provider receives the complete source-properties map and applies the same strict GTID profile as the Table sink. The serializer selects after for snapshot, create, and update operations and before for a delete. The example uses the default stream because BigQuery CDC is supported only by STORAGE_API_AT_LEAST_ONCE.

MySQL SQL sink through a DataStream bridge#

Flink’s Debezium Avro format does not currently retain the source metadata required by this profile, while the raw Avro format cannot produce the required delete row kinds through SQL alone. Issue #706 tracks that format boundary. Until it changes, register a changelog view from the complete envelope stream:

env.enableCheckpointing(60_000);
DataStream<Row> changes =
        env.fromSource(
                        kafkaSource,
                        WatermarkStrategy.noWatermarks(),
                        "Debezium MySQL orders")
                .map(
                        envelope -> {
                            GenericRecord value = currentRow(envelope);
                            return Row.ofKind(
                                    rowKind(envelope),
                                    value.get("id").toString(),
                                    value.get("amount"),
                                    sourceProperties(source(envelope)));
                        })
                .returns(
                        Types.ROW_NAMED(
                                new String[] {"id", "amount", "source_properties"},
                                Types.STRING,
                                Types.LONG,
                                Types.MAP(Types.STRING, Types.STRING)));

tableEnv.createTemporaryView(
        "mysql_source_changes",
        tableEnv.fromChangelogStream(
                changes,
                Schema.newBuilder()
                        .column("id", DataTypes.STRING().notNull())
                        .column("amount", DataTypes.BIGINT())
                        .column(
                                "source_properties",
                                DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))
                        .primaryKey("id")
                        .build(),
                ChangelogMode.upsert()));

Define the BigQuery sink table and map the source-properties column to writable metadata:

CREATE TABLE current_mysql_orders (
  id STRING NOT NULL,
  amount BIGINT,
  source_properties MAP<STRING, STRING>
    METADATA FROM 'debezium-source-properties',
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'bigquery',
  'project' = 'my-project',
  'dataset' = 'analytics',
  'table' = 'current_mysql_orders',
  'sink.cdc.enabled' = 'true',
  'sink.cdc.debezium-mysql.source-uuids' = '24bc7850-2c16-11e6-a073-0242ac110002',
  'sink.create-disposition' = 'create-if-needed',
  'sink.cdc.max-staleness' = '10 min',
  'sink.cdc.table-reconciliation' = 'reconcile'
);

Insert the changelog rows and their source metadata:

INSERT INTO current_mysql_orders
SELECT id, amount, source_properties FROM mysql_source_changes;

The source UUID option and the DataStream provider use the same append-only epoch list. After a non-interleaved failover, append the new SID with Flink’s semicolon delimiter, for example '24bc7850-2c16-11e6-a073-0242ac110002;3e11fa47-71ca-11e1-9e33-c80aa9429562'. Never edit or reorder an existing entry because the entry’s position defines its ordering epoch.

MySQL envelope adapter#

The examples use these helpers to select the current row and retain only the MySQL ordering properties:

private static final List<String> MYSQL_SOURCE_UUIDS =
        Collections.singletonList("24bc7850-2c16-11e6-a073-0242ac110002");

private static final DebeziumMySqlCdcSequenceNumberProvider MYSQL_SEQUENCE_NUMBERS =
        new DebeziumMySqlCdcSequenceNumberProvider(MYSQL_SOURCE_UUIDS);

private static GenericRecord source(GenericRecord envelope) {
    Object value = envelope.get("source");
    if (!(value instanceof GenericRecord)) {
        throw new IllegalArgumentException("Debezium change has no source record");
    }
    return (GenericRecord) value;
}

private static GenericRecord currentRow(GenericRecord envelope) {
    String operation = stringField(envelope, "op");
    String field;
    if ("d".equals(operation)) {
        field = "before";
    } else if ("c".equals(operation) || "r".equals(operation) || "u".equals(operation)) {
        field = "after";
    } else {
        throw new IllegalArgumentException(
                "Unsupported Debezium operation '" + operation + "'");
    }
    Object row = envelope.get(field);
    if (!(row instanceof GenericRecord)) {
        throw new IllegalArgumentException("Debezium change has no " + field + " row");
    }
    return (GenericRecord) row;
}

private static boolean isDelete(GenericRecord envelope) {
    return "d".equals(stringField(envelope, "op"));
}

private static RowKind rowKind(GenericRecord envelope) {
    String operation = stringField(envelope, "op");
    if ("d".equals(operation)) {
        return RowKind.DELETE;
    }
    if ("u".equals(operation)) {
        return RowKind.UPDATE_AFTER;
    }
    if ("c".equals(operation) || "r".equals(operation)) {
        return RowKind.INSERT;
    }
    throw new IllegalArgumentException("Unsupported Debezium operation '" + operation + "'");
}

private static Map<String, String> sourceProperties(GenericRecord source) {
    Map<String, String> properties = new HashMap<>(5);
    copySourceProperty(source, properties, "connector");
    copySourceProperty(source, properties, "snapshot");
    copySourceProperty(source, properties, "gtid");
    copySourceProperty(source, properties, "pos");
    copySourceProperty(source, properties, "row");
    return properties;
}

private static void copySourceProperty(
        GenericRecord source, Map<String, String> properties, String field) {
    Object value = source.get(field);
    if (value != null) {
        properties.put(field, value.toString());
    }
}

private static String stringField(GenericRecord record, String field) {
    Object value = record.get(field);
    if (!(value instanceof CharSequence)) {
        throw new IllegalArgumentException("Debezium record has no " + field);
    }
    return value.toString();
}

private static final class DebeziumEnvelopeSerializer
        extends BigQueryProtoSerializationSchema<GenericRecord> {

    private static final long serialVersionUID = 1L;

    private final AvroRecordSerializationSchema delegate;

    private DebeziumEnvelopeSerializer(AvroRecordSerializationSchema delegate) {
        this.delegate = delegate;
    }

    @Override
    public TableSchema getTableSchema(TableDestination destination) {
        return delegate.getTableSchema(destination);
    }

    @Override
    public Descriptors.Descriptor getDescriptor(TableDestination destination) {
        return delegate.getDescriptor(destination);
    }

    @Override
    public Object getSchemaFingerprint(TableDestination destination) {
        return delegate.getSchemaFingerprint(destination);
    }

    @Override
    public ByteString serialize(GenericRecord envelope) throws IOException {
        IndexedRecord row = currentRow(envelope);
        return delegate.serialize(row);
    }
}

This adapter serializes the complete before record for a delete. Configure MySQL with binlog_row_image=FULL, or replace the adapter with one that emits only the declared BigQuery primary key for deletes.

Initial snapshot rows target an empty BigQuery table. Tagged GTIDs, multi-source GTID sets, incremental snapshots, interleaved source histories, Group Replication primary changes, and Group Replication multi-primary histories are unsupported. See the MySQL topology support table before selecting this profile.

Debezium PostgreSQL CDC from Kafka#

Both examples consume Kafka values containing complete Debezium PostgreSQL Avro envelopes and write the current row to a BigQuery table with a primary key. They fail rather than use source.ts_ms when the PostgreSQL ordering metadata is absent or contradictory.

Kafka source#

The examples use Flink’s Kafka connector and ConfluentRegistryAvroDeserializationSchema.forGeneric(...) to retain the complete Debezium envelope, including before, after, op, and source. Pass the Avro schema for that complete envelope as debeziumEnvelopeSchema:

static KafkaSource<GenericRecord> kafkaSource(org.apache.avro.Schema debeziumEnvelopeSchema) {
    return KafkaSource.<GenericRecord>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("dbserver1.public.orders")
            .setGroupId("bigquery-cdc-orders")
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setValueOnlyDeserializer(
                    ConfluentRegistryAvroDeserializationSchema.forGeneric(
                            debeziumEnvelopeSchema, "http://schema-registry:8081"))
            .build();
}

For a Debezium tombstone, the Confluent Avro deserializer returns null and Kafka’s value-only wrapper emits no element. Flink checkpoint restoration takes precedence over the configured initial offsets, so a restarted job resumes the Kafka positions stored in its checkpoint. Add a Kafka connector release compatible with the application’s Flink version and flink-avro-confluent-registry to the job artifact. The compiled example uses Kafka connector 3.4.0-1.20 with Flink 1.20 and 5.0.0-2.2 with Flink 2.2 and 2.3; this repository does not ship either dependency in its connector artifacts.

DataStream API#

Pass the KafkaSource<GenericRecord> above as kafkaSource:

env.enableCheckpointing(60_000);
env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Debezium PostgreSQL orders")
        .sinkTo(
                BigQuerySink.<GenericRecord>builder()
                        .writeMethod(WriteMethod.STORAGE_API_AT_LEAST_ONCE)
                        .table(
                                TableDestination.of(
                                        "my-project", "analytics", "current_orders"))
                        .serializer(
                                new DebeziumEnvelopeSerializer(
                                        AvroRecordSerializationSchema.of(
                                                rowSchema,
                                                AvroSchemaOptions.builder()
                                                        .deriveRequiredColumns()
                                                        .build())))
                        .cdcTableOptions(
                                CdcTableOptions.builder()
                                        .primaryKeyColumns(Collections.singletonList("id"))
                                        .build())
                        .cdcTableReconciliationPolicy(
                                CdcTableReconciliationPolicy.RECONCILE)
                        .cdcOptions(
                                CdcOptions.<GenericRecord>builder(
                                                envelope ->
                                                        isDelete(envelope)
                                                                ? CdcChangeType.DELETE
                                                                : CdcChangeType.UPSERT)
                                        .sequenceNumberProvider(
                                                envelope ->
                                                        POSTGRESQL_SEQUENCE_NUMBERS
                                                                .getSequenceNumber(
                                                                        sourceProperties(
                                                                                source(
                                                                                        envelope))))
                                        .build())
                        .build());

env.execute("debezium-postgresql-to-bigquery-cdc");

The adapter selects after for snapshot, create and update operations and before for a delete. The serializer passes only that nested row to AvroRecordSerializationSchema; the two CDC providers still receive the complete envelope and derive the operation and sequence from it. The example uses the default stream because BigQuery CDC is supported only by STORAGE_API_AT_LEAST_ONCE.

SQL sink through a DataStream bridge#

NOTE: There is intentionally no Kafka source-table DDL in this example. Flink’s debezium-avro-confluent format creates the correct changelog row kinds but does not retain source.sequence or source.lsn in its output. The raw avro-confluent format retains the envelope but exposes it as insert-only rows, so SQL DDL alone cannot produce both the Debezium changelog semantics and the source metadata required by this BigQuery CDC profile. Declaring that raw envelope as a source table does not close the gap. A CASE expression can select before or after from its op value, but SQL expressions cannot change an INSERT RowKind into the DELETE RowKind required for a Debezium delete. Issue #706 tracks the upstream format improvement.

Source changelog view#

Until that is available, pass the KafkaSource<GenericRecord> above to a DataStream adapter and register its output as the source_changes changelog view on the tableEnv that will execute the sink DDL and INSERT statement:

env.enableCheckpointing(60_000);
DataStream<Row> changes =
        env.fromSource(
                        kafkaSource,
                        WatermarkStrategy.noWatermarks(),
                        "Debezium PostgreSQL orders")
                .map(
                        envelope -> {
                            GenericRecord value = currentRow(envelope);
                            return Row.ofKind(
                                    rowKind(envelope),
                                    value.get("id").toString(),
                                    value.get("amount"),
                                    sourceProperties(source(envelope)));
                        })
                .returns(
                        Types.ROW_NAMED(
                                new String[] {"id", "amount", "source_properties"},
                                Types.STRING,
                                Types.LONG,
                                Types.MAP(Types.STRING, Types.STRING)));

tableEnv.createTemporaryView(
        "source_changes",
        tableEnv.fromChangelogStream(
                changes,
                Schema.newBuilder()
                        .column("id", DataTypes.STRING().notNull())
                        .column("amount", DataTypes.BIGINT())
                        .column(
                                "source_properties",
                                DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))
                        .primaryKey("id")
                        .build(),
                ChangelogMode.upsert()));

The bridge emits the physical columns with Flink row kinds and a source-properties map. This registered view is the source relation for the remaining SQL statements.

Sink table#

Define the BigQuery sink table and map the source-properties column to writable metadata:

CREATE TABLE current_orders (
  id STRING NOT NULL,
  amount BIGINT,
  source_properties MAP<STRING, STRING>
    METADATA FROM 'debezium-source-properties',
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'bigquery',
  'project' = 'my-project',
  'dataset' = 'analytics',
  'table' = 'current_orders',
  'sink.cdc.enabled' = 'true',
  'sink.create-disposition' = 'create-if-needed',
  'sink.cdc.max-staleness' = '10 min',
  'sink.cdc.table-reconciliation' = 'reconcile'
);
Insert query#

Forward the changelog rows and their ordering metadata from the source view into the sink table:

INSERT INTO current_orders
SELECT id, amount, source_properties FROM source_changes;

The job’s checkpoint restores Kafka offsets after a Flink failure. Records between the restored offset and the last successful BigQuery append can be replayed, but the same Debezium event produces the same change sequence number.

The provider transcodes Debezium’s decimal sequence pair into the hexadecimal sections required by BigQuery; it does not parse PostgreSQL’s customary X/Y display form. PostgreSQL defines an LSN as a 64-bit WAL byte position, while Debezium defines source.sequence as the last committed LSN followed by the current LSN. Using both positions preserves transaction progress when adjacent operations share a current LSN. It does not create a unique total order for multiple events that share both positions. If distinct changes for the same BigQuery primary key can have an identical pair and can arrive out of order, provide an application-specific stable tie-breaker through the formatted change-sequence-number metadata or a custom CdcSequenceNumberProvider. See the PostgreSQL pg_lsn type, Debezium PostgreSQL source metadata, and BigQuery CDC ordering format.

Envelope adapter#

Both examples use the following adapter helpers. The DataStream path calls the connector’s DebeziumPostgreSqlCdcSequenceNumberProvider, so it applies the same strict sequence parser as the SQL sink profile rather than maintaining a second implementation in the application.

private static final DebeziumPostgreSqlCdcSequenceNumberProvider POSTGRESQL_SEQUENCE_NUMBERS =
        new DebeziumPostgreSqlCdcSequenceNumberProvider();

private static GenericRecord source(GenericRecord envelope) {
    Object value = envelope.get("source");
    if (!(value instanceof GenericRecord)) {
        throw new IllegalArgumentException("Debezium change has no source record");
    }
    return (GenericRecord) value;
}

private static GenericRecord currentRow(GenericRecord envelope) {
    String operation = stringField(envelope, "op");
    String field;
    if ("d".equals(operation)) {
        field = "before";
    } else if ("c".equals(operation) || "r".equals(operation) || "u".equals(operation)) {
        field = "after";
    } else {
        throw new IllegalArgumentException(
                "Unsupported Debezium operation '" + operation + "'");
    }
    Object row = envelope.get(field);
    if (!(row instanceof GenericRecord)) {
        throw new IllegalArgumentException("Debezium change has no " + field + " row");
    }
    return (GenericRecord) row;
}

private static boolean isDelete(GenericRecord envelope) {
    return "d".equals(stringField(envelope, "op"));
}

private static RowKind rowKind(GenericRecord envelope) {
    String operation = stringField(envelope, "op");
    if ("d".equals(operation)) {
        return RowKind.DELETE;
    }
    if ("u".equals(operation)) {
        return RowKind.UPDATE_AFTER;
    }
    if ("c".equals(operation) || "r".equals(operation)) {
        return RowKind.INSERT;
    }
    throw new IllegalArgumentException("Unsupported Debezium operation '" + operation + "'");
}

private static Map<String, String> sourceProperties(GenericRecord source) {
    Map<String, String> properties = new HashMap<>(3);
    copySourceProperty(source, properties, "connector");
    copySourceProperty(source, properties, "sequence");
    copySourceProperty(source, properties, "lsn");
    return properties;
}

private static void copySourceProperty(
        GenericRecord source, Map<String, String> properties, String field) {
    Object value = source.get(field);
    if (value != null) {
        properties.put(field, value.toString());
    }
}

private static String stringField(GenericRecord record, String field) {
    Object value = record.get(field);
    if (!(value instanceof CharSequence)) {
        throw new IllegalArgumentException("Debezium record has no " + field);
    }
    return value.toString();
}

private static final class DebeziumEnvelopeSerializer
        extends BigQueryProtoSerializationSchema<GenericRecord> {

    private static final long serialVersionUID = 1L;

    private final AvroRecordSerializationSchema delegate;

    private DebeziumEnvelopeSerializer(AvroRecordSerializationSchema delegate) {
        this.delegate = delegate;
    }

    @Override
    public TableSchema getTableSchema(TableDestination destination) {
        return delegate.getTableSchema(destination);
    }

    @Override
    public Descriptors.Descriptor getDescriptor(TableDestination destination) {
        return delegate.getDescriptor(destination);
    }

    @Override
    public Object getSchemaFingerprint(TableDestination destination) {
        return delegate.getSchemaFingerprint(destination);
    }

    @Override
    public ByteString serialize(GenericRecord envelope) throws IOException {
        IndexedRecord row = currentRow(envelope);
        return delegate.serialize(row);
    }
}

This example serializes the complete before record for a delete, so set the PostgreSQL table to REPLICA IDENTITY FULL to make that complete row image available. A different adapter that serializes only the declared BigQuery primary key for deletes can use PostgreSQL’s default replica identity when the source table has that primary key.

Both APIs require the Debezium connector to continue the same preserved or synchronized logical replication slot across a PostgreSQL primary failover. A recreated or discontinuous slot starts another LSN history with no ordering epoch, which neither example can make safe; see the BigQuery table CDC contract.

TiCDC Debezium CDC from Kafka#

TiCDC replicates TiDB row changes to Kafka in a Debezium-compatible JSON envelope when the changefeed sets protocol=debezium. Its source object carries connector=TiCDC, the transaction’s commit timestamp oracle value as commit_ts, and the reporting TiCDC cluster’s cluster_id; both fields are present from TiCDC v8.0.0 onwards. The connector’s TiCdcSequenceNumberProvider writes that commit TSO as the BigQuery change sequence, after checking that the event belongs to the configured cluster. That identifier is TiCDC’s own cluster ID, which defaults to default, so give each TiCDC cluster its own before routing more than one of them into one table.

The commit TSO orders every transaction within one TiDB cluster and survives a TiCDC process or node failover, so this profile configures no epoch list of the kind the MySQL GTID profile needs. It never falls back to the source object’s ts_ms, which truncates the same value to milliseconds and so cannot order two transactions committed within one millisecond.

Both examples below cover row changes only. Whether anything else reaches the topic depends on the TiCDC deployment. Before TiDB v9.0.0, TiCDC’s classic architecture sends row changes only in this protocol, so nothing below has to be configured against other events, while its new architecture sends DDL and watermark events from TiCDC v8.5.4-release.1. From TiDB v9.0.0 that new architecture is the only one, so DDL events always reach the topic. Watermark events additionally require the changefeed to set enable-tidb-extension=true. Neither carries a row to write, and neither is a shape Flink’s debezium-json format can deserialize: a watermark’s op value is unknown to it and a DDL event has no op field at all, so either one fails the job rather than being skipped. Where such a topic is unavoidable, set 'value.debezium-json.ignore-parse-errors' = 'true' on the source table, accepting that it also hides a genuinely malformed row change. TiCDC provides no initial snapshot stream either, so load an already-populated table separately, taking that load at the changefeed’s start timestamp.

TiCDC SQL#

Unlike the Debezium Avro sections above, this path needs no DataStream bridge. TiCDC’s Debezium protocol is JSON only; its separate avro protocol carries TiCDC’s own envelope rather than a Debezium one, so debezium-avro-confluent does not read it. Flink’s debezium-json format produces the changelog row kinds and retains the source object as value.source.properties:

CREATE TABLE source_changes (
  id STRING NOT NULL,
  amount BIGINT,
  source_properties MAP<STRING, STRING>
    METADATA FROM 'value.source.properties' VIRTUAL,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'topic' = 'tidb_test.test.orders',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'bigquery-cdc-orders',
  'scan.startup.mode' = 'earliest-offset',
  'value.format' = 'debezium-json',
  'value.debezium-json.schema-include' = 'true'
);

Keep value.debezium-json.schema-include at true whatever the changefeed sets: TiCDC always wraps the change in a payload object, which is what that option describes, while debezium-disable-schema=true removes only the sibling schema object. Reading a payload-wrapped message with schema-include set to false fails every record.

Define the BigQuery sink table with the cluster identity and the writable metadata column:

CREATE TABLE current_orders (
  id STRING NOT NULL,
  amount BIGINT,
  source_properties MAP<STRING, STRING>
    METADATA FROM 'debezium-source-properties',
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'bigquery',
  'project' = 'my-project',
  'dataset' = 'analytics',
  'table' = 'current_orders',
  'sink.cdc.enabled' = 'true',
  'sink.cdc.ticdc.cluster-id' = 'tidb-prod',
  'sink.create-disposition' = 'create-if-needed',
  'sink.cdc.max-staleness' = '10 min',
  'sink.cdc.table-reconciliation' = 'reconcile'
);

INSERT INTO current_orders
SELECT id, amount, source_properties FROM source_changes;

TiCDC Kafka source#

The DataStream example keeps the complete envelope by reading each message as text:

static KafkaSource<String> kafkaSource() {
    return KafkaSource.<String>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("tidb_test.test.orders")
            .setGroupId("bigquery-cdc-orders")
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();
}

Add a Kafka connector release compatible with the application’s Flink version to the job artifact; this repository does not ship one. The compiled example uses Kafka connector 3.4.0-1.20 with Flink 1.20 and 5.0.0-2.2 with Flink 2.2 and 2.3.

TiCDC DataStream API#

Pass the KafkaSource<String> above as kafkaSource, and the destination table’s schema as rowSchema, which is the Storage Write API’s TableSchema:

env.enableCheckpointing(60_000);
env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "TiCDC orders")
        .sinkTo(
                BigQuerySink.<String>builder()
                        .writeMethod(WriteMethod.STORAGE_API_AT_LEAST_ONCE)
                        .table(
                                TableDestination.of(
                                        "my-project", "analytics", "current_orders"))
                        .serializer(
                                new TiCdcEnvelopeSerializer(
                                        JsonDocumentSerializationSchema.of(rowSchema)))
                        .cdcTableOptions(
                                CdcTableOptions.builder()
                                        .primaryKeyColumns(Collections.singletonList("id"))
                                        .build())
                        .cdcTableReconciliationPolicy(
                                CdcTableReconciliationPolicy.RECONCILE)
                        .cdcOptions(
                                CdcOptions.<String>builder(
                                                message ->
                                                        isDelete(payload(message))
                                                                ? CdcChangeType.DELETE
                                                                : CdcChangeType.UPSERT)
                                        .sequenceNumberProvider(
                                                message ->
                                                        COMMIT_TSO_SEQUENCE_NUMBERS
                                                                .getSequenceNumber(
                                                                        sourceProperties(
                                                                                payload(
                                                                                        message))))
                                        .build())
                        .build());

env.execute("ticdc-to-bigquery-cdc");

The adapter selects after for create and update operations and before for a delete, and passes only that nested row to JsonDocumentSerializationSchema. JSON carries no schema, so that serializer takes the destination schema rather than deriving one. The two CDC providers still receive the complete message and derive the operation and sequence from it. The example uses the default stream because BigQuery CDC is supported only by STORAGE_API_AT_LEAST_ONCE.

A TiDB delete gives before the complete row image, so nothing corresponds to the PostgreSQL example’s REPLICA IDENTITY FULL for deletes. An update is different: a changefeed with sink.debezium.output-old-value=false omits before from updates, which the SQL path’s debezium-json format rejects.

TiCDC envelope adapter#

The DataStream example uses the following adapter helpers:

private static final TiCdcSequenceNumberProvider COMMIT_TSO_SEQUENCE_NUMBERS =
        new TiCdcSequenceNumberProvider("tidb-prod");

/**
 * Returns the change itself. TiCDC wraps it in a {@code payload} object unless the changefeed
 * sets {@code debezium-disable-schema=true}.
 */
private static JSONObject payload(String message) {
    JSONObject envelope = new JSONObject(message);
    return envelope.optJSONObject("payload") == null
            ? envelope
            : envelope.getJSONObject("payload");
}

private static JSONObject currentRow(JSONObject payload) {
    String field = isDelete(payload) ? "before" : "after";
    JSONObject row = payload.optJSONObject(field);
    if (row == null) {
        throw new IllegalArgumentException("TiCDC change has no " + field + " row");
    }
    return row;
}

private static boolean isDelete(JSONObject payload) {
    String operation = payload.optString("op", null);
    if ("c".equals(operation) || "u".equals(operation) || "r".equals(operation)) {
        return false;
    }
    if ("d".equals(operation)) {
        return true;
    }
    // A DDL or watermark event carries no row to write; only TiCDC's new architecture
    // emits them.
    throw new IllegalArgumentException("Unsupported TiCDC operation '" + operation + "'");
}

private static Map<String, String> sourceProperties(JSONObject payload) {
    JSONObject source = payload.optJSONObject("source");
    if (source == null) {
        throw new IllegalArgumentException("TiCDC change has no source object");
    }
    Map<String, String> properties = new HashMap<>(4);
    copySourceProperty(source, properties, "connector");
    copySourceProperty(source, properties, "snapshot");
    copySourceProperty(source, properties, "commit_ts");
    copySourceProperty(source, properties, "cluster_id");
    return properties;
}

private static void copySourceProperty(
        JSONObject source, Map<String, String> properties, String field) {
    if (!source.isNull(field)) {
        properties.put(field, String.valueOf(source.get(field)));
    }
}

private static final class TiCdcEnvelopeSerializer
        extends BigQueryProtoSerializationSchema<String> {

    private static final long serialVersionUID = 1L;

    private final JsonDocumentSerializationSchema delegate;

    private TiCdcEnvelopeSerializer(JsonDocumentSerializationSchema delegate) {
        this.delegate = delegate;
    }

    @Override
    public TableSchema getTableSchema(TableDestination destination) {
        return delegate.getTableSchema(destination);
    }

    @Override
    public Descriptors.Descriptor getDescriptor(TableDestination destination) {
        return delegate.getDescriptor(destination);
    }

    @Override
    public Object getSchemaFingerprint(TableDestination destination) {
        return delegate.getSchemaFingerprint(destination);
    }

    @Override
    public ByteString serialize(String message) throws IOException {
        return delegate.serialize(currentRow(payload(message)).toString());
    }
}

The job’s checkpoint restores Kafka offsets after a Flink failure. Records between the restored offset and the last successful BigQuery append can be replayed, but the same TiCDC event produces the same change sequence number.

Every row change of one transaction carries that transaction’s commit TSO and therefore one sequence, and a later transaction always supersedes an earlier one.

NOTE: One transaction can still produce two conflicting changes for one BigQuery primary key. TiDB writes each key at most once per transaction, but TiCDC splits an UPDATE that modifies a primary or unique key into a DELETE and an INSERT, which is the default for every sink except MySQL. A transaction that moves one key’s value onto another key, such as the primary-key swap in TiCDC’s UPDATE splitting behavior, emits both a DELETE and an INSERT for one key at one sequence, and BigQuery resolves that pair by ingestion time rather than by TiCDC’s emission order. Where such transactions occur, supply an application tie-breaker through change-sequence-number or a custom CdcSequenceNumberProvider; see the BigQuery table CDC contract.

See the TiCDC Debezium protocol, TiDB timestamp oracle, and BigQuery CDC ordering format.

Spanner CDC from either route#

Spanner changes reach BigQuery either through a Debezium Spanner connector on Kafka or through this repository’s native Change Streams source. Both routes encode the same three coordinates of one Spanner mod — the commit timestamp in nanoseconds, the record sequence within the transaction, and the mod number within the record — so the same change produces the same BigQuery sequence whichever route wrote it.

Debezium Spanner Kafka source#

Use Flink’s Kafka connector and ConfluentRegistryAvroDeserializationSchema.forGeneric(...) to retain the complete envelope. Pass its Avro schema as debeziumEnvelopeSchema:

static KafkaSource<GenericRecord> kafkaSource(org.apache.avro.Schema debeziumEnvelopeSchema) {
    return KafkaSource.<GenericRecord>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("my-instance.my-database.Orders")
            .setGroupId("bigquery-cdc-orders")
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setValueOnlyDeserializer(
                    ConfluentRegistryAvroDeserializationSchema.forGeneric(
                            debeziumEnvelopeSchema, "http://schema-registry:8081"))
            .build();
}

Debezium Spanner DataStream API#

DebeziumSpannerCdcSequenceNumberProvider reads connector, ts_ns, sequence, and mod_number from the envelope’s source record:

env.enableCheckpointing(60_000);
env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Debezium Spanner orders")
        .sinkTo(
                BigQuerySink.<GenericRecord>builder()
                        .writeMethod(WriteMethod.STORAGE_API_AT_LEAST_ONCE)
                        .table(
                                TableDestination.of(
                                        "my-project", "analytics", "current_orders"))
                        .serializer(
                                new DebeziumEnvelopeSerializer(
                                        AvroRecordSerializationSchema.of(
                                                rowSchema,
                                                AvroSchemaOptions.builder()
                                                        .deriveRequiredColumns()
                                                        .build())))
                        .cdcTableOptions(
                                CdcTableOptions.builder()
                                        .primaryKeyColumns(Collections.singletonList("id"))
                                        .build())
                        .cdcTableReconciliationPolicy(
                                CdcTableReconciliationPolicy.RECONCILE)
                        .cdcOptions(
                                CdcOptions.<GenericRecord>builder(
                                                envelope ->
                                                        isDelete(envelope)
                                                                ? CdcChangeType.DELETE
                                                                : CdcChangeType.UPSERT)
                                        .sequenceNumberProvider(
                                                envelope ->
                                                        SPANNER_SEQUENCE_NUMBERS
                                                                .getSequenceNumber(
                                                                        sourceProperties(
                                                                                source(
                                                                                        envelope))))
                                        .build())
                        .build());

env.execute("debezium-spanner-to-bigquery-cdc");

Debezium Spanner SQL sink through a DataStream bridge#

Registering the changelog as a view keeps the source properties available to SQL:

env.enableCheckpointing(60_000);
DataStream<Row> changes =
        env.fromSource(
                        kafkaSource,
                        WatermarkStrategy.noWatermarks(),
                        "Debezium Spanner orders")
                .map(
                        envelope -> {
                            GenericRecord value = currentRow(envelope);
                            return Row.ofKind(
                                    rowKind(envelope),
                                    value.get("id").toString(),
                                    value.get("amount"),
                                    sourceProperties(source(envelope)));
                        })
                .returns(
                        Types.ROW_NAMED(
                                new String[] {"id", "amount", "source_properties"},
                                Types.STRING,
                                Types.LONG,
                                Types.MAP(Types.STRING, Types.STRING)));

tableEnv.createTemporaryView(
        "source_changes",
        tableEnv.fromChangelogStream(
                changes,
                Schema.newBuilder()
                        .column("id", DataTypes.STRING().notNull())
                        .column("amount", DataTypes.BIGINT())
                        .column(
                                "source_properties",
                                DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))
                        .primaryKey("id")
                        .build(),
                ChangelogMode.upsert()));
CREATE TABLE current_orders (
  id STRING NOT NULL,
  amount BIGINT,
  source_properties MAP<STRING, STRING> METADATA FROM 'debezium-source-properties',
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'bigquery',
  'project' = 'my-project',
  'dataset' = 'analytics',
  'table' = 'current_orders',
  'sink.cdc.enabled' = 'true',
  'sink.create-disposition' = 'create-if-needed',
  'sink.cdc.max-staleness' = '10 min'
);

INSERT INTO current_orders
SELECT id, amount, source_properties FROM source_changes;

Debezium Spanner envelope adapter#

The adapter turns one envelope into the row to write and the source properties to order it by:

private static final DebeziumSpannerCdcSequenceNumberProvider SPANNER_SEQUENCE_NUMBERS =
        new DebeziumSpannerCdcSequenceNumberProvider();

private static GenericRecord source(GenericRecord envelope) {
    Object value = envelope.get("source");
    if (!(value instanceof GenericRecord)) {
        throw new IllegalArgumentException("Debezium change has no source record");
    }
    return (GenericRecord) value;
}

private static GenericRecord currentRow(GenericRecord envelope) {
    String operation = stringField(envelope, "op");
    String field;
    if ("d".equals(operation)) {
        field = "before";
    } else if ("c".equals(operation) || "r".equals(operation) || "u".equals(operation)) {
        field = "after";
    } else {
        throw new IllegalArgumentException(
                "Unsupported Debezium operation '" + operation + "'");
    }
    Object row = envelope.get(field);
    if (!(row instanceof GenericRecord)) {
        throw new IllegalArgumentException("Debezium change has no " + field + " row");
    }
    return (GenericRecord) row;
}

private static boolean isDelete(GenericRecord envelope) {
    return "d".equals(stringField(envelope, "op"));
}

private static RowKind rowKind(GenericRecord envelope) {
    String operation = stringField(envelope, "op");
    if ("d".equals(operation)) {
        return RowKind.DELETE;
    }
    if ("u".equals(operation)) {
        return RowKind.UPDATE_AFTER;
    }
    if ("c".equals(operation) || "r".equals(operation)) {
        return RowKind.INSERT;
    }
    throw new IllegalArgumentException("Unsupported Debezium operation '" + operation + "'");
}

/**
 * Reads the ordering coordinates from the envelope's {@code source} record. The sibling {@code
 * ts_ns} directly under the payload is the connector's processing time and must not be copied
 * here.
 */
private static Map<String, String> sourceProperties(GenericRecord source) {
    Map<String, String> properties = new HashMap<>(4);
    copySourceProperty(source, properties, "connector");
    copySourceProperty(source, properties, "ts_ns");
    copySourceProperty(source, properties, "sequence");
    copySourceProperty(source, properties, "mod_number");
    return properties;
}

private static void copySourceProperty(
        GenericRecord source, Map<String, String> properties, String field) {
    Object value = source.get(field);
    if (value != null) {
        properties.put(field, value.toString());
    }
}

private static String stringField(GenericRecord record, String field) {
    Object value = record.get(field);
    if (!(value instanceof CharSequence)) {
        throw new IllegalArgumentException("Debezium record has no " + field);
    }
    return value.toString();
}

private static final class DebeziumEnvelopeSerializer
        extends BigQueryProtoSerializationSchema<GenericRecord> {

    private static final long serialVersionUID = 1L;

    private final AvroRecordSerializationSchema delegate;

    private DebeziumEnvelopeSerializer(AvroRecordSerializationSchema delegate) {
        this.delegate = delegate;
    }

    @Override
    public TableSchema getTableSchema(TableDestination destination) {
        return delegate.getTableSchema(destination);
    }

    @Override
    public Descriptors.Descriptor getDescriptor(TableDestination destination) {
        return delegate.getDescriptor(destination);
    }

    @Override
    public Object getSchemaFingerprint(TableDestination destination) {
        return delegate.getSchemaFingerprint(destination);
    }

    @Override
    public ByteString serialize(GenericRecord envelope) throws IOException {
        IndexedRecord row = currentRow(envelope);
        return delegate.serialize(row);
    }
}

Copy the coordinates from the envelope’s source record rather than from the payload. Debezium writes ts_ms, ts_us, and ts_ns in both places: inside source they carry the Spanner commit timestamp, while the siblings beside source carry the time the connector processed the event on its own clock.

The adapter rejects every operation it does not recognize, which includes the m of the low-watermark stamps Debezium writes into this same topic under gcp.spanner.low-watermark.enabled. Those stamps carry no row and no ordering coordinates, so there is nothing to write and nothing to order them by; leave that option at its default false for a topic this example reads.

Native Change Streams DataStream API#

The native source hands the deserializer a typed DataChangeRecord, so the example emits one element per mod and keeps the mod’s position as its mod number:


/**
 * Emits one element per mod. The mod number is the mod's zero-based position in the change
 * record, which is what makes several mods of one record mutually orderable.
 */
static final class OrderModDeserializer
        implements SpannerChangeStreamDeserializationSchema<OrderMod> {

    private static final long serialVersionUID = 1L;

    @Override
    public void deserialize(DataChangeRecord record, Collector<OrderMod> out)
            throws IOException {
        List<Mod> mods = record.getMods();
        for (int modNumber = 0; modNumber < mods.size(); modNumber++) {
            Mod mod = mods.get(modNumber);
            boolean deletion = record.getModType() == ModType.DELETE;
            out.collect(
                    new OrderMod(
                            record.getCommitTimestamp(),
                            record.getRecordSequence(),
                            modNumber,
                            deletion,
                            deletion
                                    ? mod.getKeysJson()
                                    : mod.getNewValuesJson()
                                            .orElseThrow(
                                                    () ->
                                                            new IOException(
                                                                    "The change stream captures no new"
                                                                            + " values for this mod"))));
        }
    }

    @Override
    public TypeInformation<OrderMod> getProducedType() {
        return TypeInformation.of(OrderMod.class);
    }
}

/** One Spanner mod with the three coordinates BigQuery orders it by. */
static final class OrderMod implements Serializable {

    private static final long serialVersionUID = 1L;

    private final Instant commitTimestamp;
    private final String recordSequence;
    private final int modNumber;
    private final boolean deletion;
    private final String rowJson;

    OrderMod(
            Instant commitTimestamp,
            String recordSequence,
            int modNumber,
            boolean deletion,
            String rowJson) {
        this.commitTimestamp = commitTimestamp;
        this.recordSequence = recordSequence;
        this.modNumber = modNumber;
        this.deletion = deletion;
        this.rowJson = rowJson;
    }

    Instant commitTimestamp() {
        return commitTimestamp;
    }

    String recordSequence() {
        return recordSequence;
    }

    int modNumber() {
        return modNumber;
    }

    boolean isDeletion() {
        return deletion;
    }

    String rowJson() {
        return rowJson;
    }
}

SpannerCdcSequenceNumber.of(...) then encodes those coordinates without going through a Debezium-shaped map:

env.enableCheckpointing(60_000);
SpannerChangeStreamSource<OrderMod> source =
        SpannerChangeStreamSource.<OrderMod>builder()
                .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
                .changeStreamName("order_changes")
                .deserializer(new OrderModDeserializer())
                .startPosition(StartPosition.latest())
                .build();

env.fromSource(source, WatermarkStrategy.noWatermarks(), "spanner-order-changes")
        .sinkTo(
                BigQuerySink.<OrderMod>builder()
                        .writeMethod(WriteMethod.STORAGE_API_AT_LEAST_ONCE)
                        .table(
                                TableDestination.of(
                                        "my-project", "analytics", "current_orders"))
                        .serializer(
                                new OrderModSerializer(
                                        JsonDocumentSerializationSchema.of(rowSchema)))
                        .cdcTableOptions(
                                CdcTableOptions.builder()
                                        .primaryKeyColumns(
                                                Collections.singletonList("OrderId"))
                                        .build())
                        .cdcOptions(
                                CdcOptions.<OrderMod>builder(
                                                mod ->
                                                        mod.isDeletion()
                                                                ? CdcChangeType.DELETE
                                                                : CdcChangeType.UPSERT)
                                        .sequenceNumberProvider(
                                                mod ->
                                                        SpannerCdcSequenceNumber.of(
                                                                mod.commitTimestamp(),
                                                                mod.recordSequence(),
                                                                mod.modNumber()))
                                        .build())
                        .build());

env.execute("spanner-change-stream-to-bigquery-cdc");

Native Change Streams SQL#

The native route needs no DataStream bridge. The Debezium route needs one because Flink’s Avro changelog drops the envelope’s source record, whereas the Spanner source of this repository is a Flink table in its own right and already exposes its ordering coordinates as metadata columns. Source DDL, sink DDL, and the INSERT are therefore all the SQL below.

Declare the change stream as a source table and expose its three ordering coordinates as metadata columns:

SET 'execution.checkpointing.interval' = '1 min';

CREATE TABLE order_changes (
  OrderId BIGINT,
  Customer STRING,
  Status STRING,
  commit_timestamp TIMESTAMP_LTZ(9) METADATA FROM 'commit-timestamp' VIRTUAL,
  record_sequence STRING METADATA FROM 'sequence' VIRTUAL,
  mod_number INT METADATA FROM 'mod-number' VIRTUAL,
  PRIMARY KEY (OrderId) NOT ENFORCED
) WITH (
  'connector' = 'spanner',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'database' = 'orders-db',
  'table' = 'Orders',
  'scan.mode' = 'change-stream',
  'scan.change-stream.name' = 'order_changes',
  'scan.change-stream.changelog-mode' = 'upsert',
  'scan.startup.mode' = 'latest'
);

Declare the BigQuery table and take the sequence as one row of writable metadata:

CREATE TABLE current_orders (
  OrderId BIGINT NOT NULL,
  Customer STRING,
  Status STRING,
  change_sequence ROW<commit_timestamp TIMESTAMP_LTZ(9), record_sequence STRING, mod_number INT>
    METADATA FROM 'spanner-change-sequence',
  PRIMARY KEY (OrderId) NOT ENFORCED
) WITH (
  'connector' = 'bigquery',
  'project' = 'my-project',
  'dataset' = 'analytics',
  'table' = 'current_orders',
  'sink.cdc.enabled' = 'true',
  'sink.create-disposition' = 'create-if-needed',
  'sink.cdc.max-staleness' = '10 min'
);

Insert the changelog and build the sequence row from the three source metadata columns:

INSERT INTO current_orders
SELECT OrderId, Customer, Status,
       ROW(commit_timestamp, record_sequence, mod_number)
FROM order_changes;

Four details in and behind that DDL are load-bearing. The checkpoint interval persists the Change Streams position and flushes the BigQuery default stream, so it must remain enabled for at-least-once recovery. The source runs in upsert changelog mode, because full also emits update-before rows, which the BigQuery CDC sink rejects. The physical order_changes stream must already exist with value_capture_type set to NEW_ROW or NEW_ROW_AND_OLD_VALUES; OLD_AND_NEW_VALUES cannot supply the complete after-image that upsert mode requires. The Spanner Change Streams setup shows the physical stream creation step. The commit-timestamp column is declared TIMESTAMP_LTZ(9) and carries no watermark: Flink permits watermark columns only through precision 3, and truncating the commit timestamp to milliseconds would let two changes of one key inside the same millisecond compare equal on their first section.

This is an analytics-replica pipeline: it keeps the current row shape in BigQuery for analytical queries, rather than promising a byte-for-byte or transactionally consistent Spanner replica. The source DDL starts at latest, so a fresh job materializes only mutations committed after it starts. A populated source needs a separate initial snapshot or backfill coordinated with a timestamp Change Streams handoff before the BigQuery table contains every existing key; that bootstrap is outside this example. The pipeline is still at-least-once, but the spanner-change-sequence metadata prevents BigQuery from resolving ordered changes for one key merely by append arrival time. Those three coordinates do not create a total order where Spanner exposes none: disjoint column writes can retain equal coordinates, which BigQuery still resolves by ingestion order.

The Bigtable selected-cell analytics replica shows the corresponding keyed upsert/delete shape for a single Bigtable cell. That example does not supply BigQuery sequence metadata, so conflicting changes for one key are resolved by BigQuery ingestion order rather than source coordinates.

Local development#

Pointing the sink at an emulator#

The sink takes two emulator endpoints, one per transport, because BigQuery serves table metadata over REST and the Storage Write API over gRPC — where every sibling connector needs one transport and so exposes one endpoint. Against goccy/bigquery-emulator, emulatorEndpoint(...) points the Storage Write API traffic at it; emulatorRestEndpoint(...) points the table metadata traffic at it, which means table creation under CREATE_IF_NEEDED, connector-driven schema updates and the CDC table contract. A sink that only appends to a table that already exists needs the first alone. Any other sink — one that creates a table, evolves a schema or manages a CDC table — needs both: given only emulatorEndpoint(...), its metadata half still reaches real BigQuery, under ADC and without saying so, so a run meant for local development can create or alter production tables.

Both are rejected under FILE_LOADS, which stages files to Cloud Storage that no emulator here stands in for: an endpoint could only be honored by the metadata half of that write method and silently ignored by the half that moves the rows. Neither can be combined with serviceAccountKeyFile(...), because emulator connections are deliberately credential-free.

What such a run proves is bounded, and worth knowing before leaning on it. The emulator reads TIME, DATETIME, NUMERIC and BIGNUMERIC columns back as unrelated values, and it keeps no flush cursor — so the exactly-once guarantee is not observable there, which is why this module’s exactly-once integration tests run against a real dataset. A sandbox project with a short default table expiration keeps that cheap.

Backward Dynamic destinations Cloud Pub/Sub Forward