Bigtable examples#

The Bigtable quickstart owns the basic source and sink jobs. The cases below change one direction or compose it with another connector.

The async SQL functions return conditional outcomes and changed cells to Flink 2.x streaming queries, with SQL-only registration and named request settings.

DataStream source#

The Quickstart read job is the canonical bounded source example. The worked cases below narrow its rows, filter its cells, or route it through another application profile.

Reading a key range#

A prefix and an explicit range are the same thing said two ways, and both are repeatable:

BigtableSource.<Order>builder()
        .table(TableDestination.of("my-project", "my-instance", "orders"))
        .deserializer(new OrderRows())
        // Everything under one prefix, plus one range named outright. Overlapping ranges
        // are
        // merged rather than rejected, so nested prefixes cost nothing but are not read
        // twice.
        .prefix("2026-08-")
        .rowRange("archive#2025-", "archive#2026-")
        .build();

What a checkpoint carries is the range that is left: after emitting the row 2026-08-14#9, the split covers (2026-08-14#9, 2026-09-). That is what makes a restore resume rather than replay, and it is why the source needs no offset of its own.

Filtering on the server#

What a filter excludes never leaves Bigtable, so it is the cheapest thing a scan can carry. Every per-cell decision belongs in that filter because the source has no separate knobs for families, qualifiers, timestamps, or versions:

BigtableSource.<Order>builder()
        .table(TableDestination.of("my-project", "my-instance", "orders"))
        .deserializer(new OrderRows())
        .filter(
                Filters.FILTERS
                        .chain()
                        .filter(Filters.FILTERS.family().exactMatch("cf"))
                        .filter(Filters.FILTERS.qualifier().exactMatch("payload"))
                        // The latest version of each cell only.
                        .filter(Filters.FILTERS.limit().cellsPerColumn(1)))
        .build();

Reading through an application profile#

BigtableSource.<Order>builder()
        .table(TableDestination.of("my-project", "my-instance", "orders"))
        .deserializer(new OrderRows())
        .appProfileId("analytics")
        .build();

A Data Boost profile is named here like any other. Its reads can be up to about 35 minutes stale, so a job that writes with the sink and reads back through it may not see its own recent writes. The profile is read-only, so using it on the sink breaks writes. This project has not exercised Data Boost; #248 owns that verification.

DataStream sink#

The Quickstart write job is the canonical fixed-table sink example. The worked cases below change the mutation, destination, rejection policy, or memory bounds.

Several cells, and a delete, per record#

One RowMutationEntry may carry up to 100,000 mutations, and they apply to the row atomically — which is the one atomicity guarantee Bigtable offers, and the reason a record that updates several columns should build one entry rather than be split upstream:

BigtableSink.<OrderEvent>builder()
        .table(TableDestination.of("my-project", "my-instance", "orders"))
        .serializer(
                (event, context) -> {
                    long timestampMicros = event.updatedAtMillis() * 1_000;
                    RowMutationEntry entry = RowMutationEntry.create("order#" + event.id());
                    entry.setCell("cf", "status", timestampMicros, event.status());
                    entry.setCell("cf", "total", timestampMicros, event.totalCents());
                    if (event.isCancelled()) {
                        // Applied together with the two cells above, in one atomic
                        // mutation.
                        entry.deleteCells("cf", "reserved_stock");
                    }
                    return entry;
                })
        .build();

Updating aggregate cells#

Bigtable can combine inputs inside an aggregate column family. The first example opts into creation of counters with an Int64 Sum family named totals and retains two bucket versions. An existing totals family must have the same aggregate type; its GC rule is left unchanged. The merge example below uses that same provisioned table.

CounterUpdate carries a row key, a bucket start in epoch milliseconds, and a signed delta. All updates for one counter bucket use the same row key, qualifier and bucket timestamp, so they address the same aggregate cell:

BigtableSink.<CounterUpdate>builder()
        .createDisposition(CreateDisposition.CREATE_IF_NEEDED)
        .tableCreateOptions(
                TableCreateOptions.builder()
                        .columnFamily(
                                "totals", ColumnFamilyType.INT64_SUM, GcRule.maxVersions(2))
                        .build())
        .table(TableDestination.of("my-project", "my-instance", "counters"))
        .serializer(
                (update, context) ->
                        RowMutationEntry.create(update.key())
                                .addToCell(
                                        "totals",
                                        "count",
                                        update.bucketStartMillis() * 1_000L,
                                        update.delta()))
        .build();

addToCell contributes an input to the family’s aggregation function. MergeToCell contributes an already accumulated state, such as the bytes read from an Int64 Sum cell in another table. The pinned Java SDK 2.82.0’s mergeToCell convenience overload encodes that state as raw_value. Its typed Value model has no bytes_value variant either. An Int64 Sum write to real Bigtable on 2026-09-05 rejected that input with INVALID_ARGUMENT: ... must use bytes_value; ADR-0041 records the observation and the successful bytes_value rerun. The example therefore builds the protobuf input with bytes_value and wraps the mutation through the SDK’s public beta fromProtoUnsafe and createFromMutationUnsafe methods. These methods bypass the mutation builder’s 200 MiB byte-size guard and permit server-side timestamps on subsequently chained setCell calls. They must not be used to infer a size bound or retry idempotence; this example supplies one mutation with an explicit timestamp and a service-produced Int64 Sum accumulator. CounterState carries the destination key, bucket start and those accumulator bytes; the state must match the destination family’s aggregate type.

Mutation and Value below are the protobuf types from com.google.bigtable.v2.

BigtableSink.<CounterState>builder()
        .table(TableDestination.of("my-project", "my-instance", "counters"))
        .serializer(
                (state, context) -> {
                    Value qualifier =
                            Value.newBuilder()
                                    .setRawValue(ByteString.copyFromUtf8("count"))
                                    .build();
                    Value timestamp =
                            Value.newBuilder()
                                    .setRawTimestampMicros(
                                            state.bucketStartMillis() * 1_000L)
                                    .build();
                    Value input =
                            Value.newBuilder().setBytesValue(state.accumulator()).build();
                    Mutation merge =
                            Mutation.newBuilder()
                                    .setMergeToCell(
                                            Mutation.MergeToCell.newBuilder()
                                                    .setFamilyName("totals")
                                                    .setColumnQualifier(qualifier)
                                                    .setTimestamp(timestamp)
                                                    .setInput(input))
                                    .build();
                    return RowMutationEntry.createFromMutationUnsafe(
                            ByteString.copyFromUtf8(state.key()),
                            com.google.cloud.bigtable.data.v2.models.Mutation
                                    .fromProtoUnsafe(List.of(merge)));
                })
        .build();

A stable bucket timestamp selects the same cell after replay; it does not identify a unique input or prevent a Sum input from being added again. The sink remains at-least-once, so use these examples only when the application can tolerate repeated contributions or prevents them outside this sink. See replay semantics.

A Flink aggregation that emits ordinary setCell upserts has a different meaning: it writes computed totals, while these operations ask Bigtable to combine inputs or states. Do not feed repeatedly emitted running totals into addToCell as deltas; each total would contribute again. The bounded SQL example uses ordinary cells.

Replacing a column immediately#

A delete-then-write removes all versions of one column and writes its replacement in one atomic row entry. Create the orders table with the ordinary family cf first, as in the Quickstart. The deletion must precede setCell in the same RowMutationEntry:

BigtableSink.<OrderEvent>builder()
        .table(TableDestination.of("my-project", "my-instance", "orders"))
        .serializer(
                (event, context) ->
                        RowMutationEntry.create("order#" + event.id())
                                .deleteCells(
                                        "cf",
                                        ByteString.copyFromUtf8("status"),
                                        TimestampRange.unbounded())
                                .setCell(
                                        "cf",
                                        "status",
                                        event.updatedAtMillis() * 1_000L,
                                        event.status()))
        .build();

After this entry succeeds, cf:status contains the replacement version and other columns are unchanged. A maxVersions(1) garbage-collection policy alone does not give that immediate result because garbage collection runs later. Other writers can subsequently add versions again.

Separate entries remain unordered, including entries for the same row and entries in different concurrent requests. An older event replayed after a newer replacement can delete that newer value, even if both carry explicit timestamps. This pattern keeps one version at the time it is applied; applications requiring event-order winners must separate dependent writes so each completes before the next begins, or encode versions in their row keys.

A table per day, from the record#

The dynamic destinations guide defines the shared resolver contract and compares the batcher’s lifetime with the other sinks. The resolver names the table, while the serializer still builds the whole mutation. The resolver runs once per record, so this map avoids reconstructing equal TableDestination values for repeated days. Equality rather than object identity keys the writer’s batcher pool, so the cache is an allocation optimization rather than a correctness requirement.

Map<LocalDate, TableDestination> byDay = new HashMap<>();

BigtableSink.<OrderEvent>builder()
        .destinationResolver(
                (event, context) ->
                        byDay.computeIfAbsent(
                                event.day(),
                                day ->
                                        TableDestination.of(
                                                "my-project",
                                                "my-instance",
                                                "orders-" + day)))
        .serializer(
                (event, context) ->
                        RowMutationEntry.create(event.id())
                                .setCell(
                                        "cf",
                                        "payload",
                                        event.timestampMicros(),
                                        event.body()))
        // A day's table stops receiving records once the day rolls over, and its batcher
        // goes with
        // it after this long. One hour is the default; this job knows its tables turn over
        // faster.
        .writerOptions(
                BigtableWriterOptions.builder()
                        .destinationIdleTimeout(Duration.ofMinutes(15))
                        .build())
        .build();

The map is captured by the resolver’s closure, so it has to reach the task manager: a HashMap built where the job is assembled travels fine, while an instance field of a class that is not serializable does not. Idle eviction closes the writer’s batcher but does not remove entries from this resolver-owned map, so the example retains one entry per observed day. Every table the resolver can name must already exist unless the sink is opted into auto-creation. Read that section beside a resolver because one schema serves every table the sink creates, and a resolver keyed on something unbounded creates one table per value.

Skipping records instead of filtering upstream#

Returning null writes nothing and is not a failure, so a filter whose condition is only known while building the mutation belongs in the serializer:

BigtableSink.<Event>builder()
        .table(TableDestination.of("my-project", "my-instance", "readings"))
        .serializer(
                (event, context) ->
                        event.isHeartbeat()
                                ? null
                                : RowMutationEntry.create("device#" + event.deviceId())
                                        .setCell(
                                                "cf",
                                                "reading",
                                                event.timestampMicros(),
                                                event.value()))
        .build();

Dropping bad rows instead of failing the job#

Only two failures are droppable — a record the serializer rejects, and a mutation the service rejects as invalid. Everything else, an outage or a missing column family included, still fails the job; the Bigtable connector page sets out why that line is where it is.

BigtableSink.<OrderEvent>builder()
        .table(TableDestination.of("my-project", "my-instance", "orders"))
        .serializer(new OrderEventMutations())
        .failedMutationHandler(FailureHandler.logAndDrop())
        .build();

A dead-letter destination is the same setter with FailureHandler.sendToDeadLetterQueue(...). What arrives there is the serialized MutateRowsRequest.Entry, so a consumer replays the whole row mutation rather than reconstructing it from a row key.

Bounding memory on large mutations#

The default byte bound is 64 MiB of unacknowledged entries. A pipeline whose rows are large — or one running many subtasks per TaskManager — sets it explicitly, and lowering the batch element count shortens the delay before a mutation reaches the service at low volume:

BigtableSink.<OrderEvent>builder()
        .table(TableDestination.of("my-project", "my-instance", "orders"))
        .serializer(new OrderEventMutations())
        .writerOptions(
                BigtableWriterOptions.builder()
                        .maxInFlightBytes(16L * 1024 * 1024)
                        .batchElementCountThreshold(50)
                        .build())
        .build();

Raising maxInFlightEntries well past its default is the one direction that does not help: the client’s own flow controller then becomes the binding limit, and it blocks the task thread rather than yielding to the mailbox — see Tuning.

Table source#

Scanning a Bigtable table with SQL#

A bounded scan treats the row key as one column and every selected column family as a nested row. Only the families selected by the query leave Bigtable. The quickstart creates the instance but not this table, so create the physical table and its families first:

cbt -project my-project -instance my-instance createtable profiles families=profile,usage,state

CREATE TABLE profiles (
  rowkey STRING,
  profile ROW<name STRING, email STRING>,
  usage ROW<requests BIGINT, last_seen TIMESTAMP_LTZ(3)>,
  PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'profiles'
);

SELECT rowkey, profile.name, usage.last_seen
FROM profiles
WHERE rowkey >= 'customer#1000' AND rowkey < 'customer#2000';

The row-key bounds are pushed into the scan. The family projection is pushed down separately. Selecting any field from profile reads that whole family; the connector discards undeclared or unselected qualifiers after the row arrives.

Table sink#

SQL aggregate contributions#

This sink contributes each input to four Bigtable aggregate families at an explicit bucket timestamp. All four columns take integers, including the HLL input. Repeated row keys remain separate INSERT contributions; the primary key does not deduplicate them.

CREATE TABLE aggregate_inputs (
    rowkey STRING,
    totals ROW<q BIGINT>,
    minimums ROW<q BIGINT>,
    maximums ROW<q BIGINT>,
    users ROW<q BIGINT>,
    bucket_ts TIMESTAMP_LTZ(6) METADATA FROM 'timestamp',
    PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
    'connector' = 'bigtable',
    'project' = 'my-project',
    'instance' = 'my-instance',
    'table' = 'aggregate-counters',
    'sink.write-mode' = 'aggregate',
    'sink.aggregate.column-family-types' = 'totals:int64-sum,minimums:int64-min,maximums:int64-max,users:int64-hll',
    'sink.create-disposition' = 'create-if-needed',
    'sink.table-create.gc-rule.max-versions' = '2'
);

INSERT INTO aggregate_inputs
SELECT 'account-a', ROW(n), ROW(n), ROW(n), ROW(n),
       TO_TIMESTAMP_LTZ(1788652800000, 3)
FROM (VALUES (CAST(3 AS BIGINT)), (CAST(5 AS BIGINT)), (CAST(3 AS BIGINT))) AS inputs(n);

After these three contributions, SUM is 11, MIN is 3, MAX is 5, and HLL represents the distinct integer inputs 3 and 5. Submitting the same inputs again at the same timestamp doubles SUM; the other three aggregate results remain unchanged unless GC or deletion removed state. Omitting the metadata column or supplying null uses the writer clock per written cell, which can produce different versions rather than one shared bucket. Supply event bucket timestamps when contributions must combine, and account for at-least-once replay.

The input DDL is sink-only because aggregate input and stored state have different types. Use a separate read DDL with numeric states declared as BIGINT and HLL state as BYTES:

CREATE TABLE aggregate_state (
    rowkey STRING,
    totals ROW<q BIGINT>,
    minimums ROW<q BIGINT>,
    maximums ROW<q BIGINT>,
    users ROW<q BYTES>
) WITH (
    'connector' = 'bigtable',
    'project' = 'my-project',
    'instance' = 'my-instance',
    'table' = 'aggregate-counters'
);

SELECT rowkey, totals.q, minimums.q, maximums.q, users.q FROM aggregate_state;

The read returns HLL sketch bytes, not cardinality. Use Bigtable SQL’s HLL_COUNT.EXTRACT to estimate the distinct count. See the aggregate mode contract for nulls, validation, provisioning permissions, and accepted input types.

Writing a bounded aggregate as upserts#

Bigtable writes are row-key upserts by construction. Declaring rowkey as the primary key tells the planner that the query and destination use the same upsert key. For a bounded aggregation, the planner can emit one final total per key and the sink overwrites the cells for that row key.

Create the destination table before submitting the statement:

cbt -project my-project -instance my-instance createtable profile-totals families=stats

SET 'execution.runtime-mode' = 'batch';

CREATE TABLE order_events (
  user_id STRING,
  amount BIGINT
) WITH (
  'connector' = 'datagen',
  'number-of-rows' = '1000'
);

CREATE TABLE profile_totals (
  rowkey STRING,
  stats ROW<order_count BIGINT, total_amount BIGINT>,
  PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'profile-totals'
);

INSERT INTO profile_totals
SELECT user_id, ROW(COUNT(*), SUM(amount))
FROM order_events
GROUP BY user_id;

The aggregate’s upsert key is user_id, which maps directly to the sink’s rowkey, so Flink 2.3 can plan this statement without an ON CONFLICT clause. Flink 2.3 requires an ON CONFLICT DO DEDUPLICATE, DO ERROR, or DO NOTHING strategy when the query key differs from the sink key or the planner cannot infer one.

The default sink.insert-only-input-mode = upsert exposes those conflict strategies for an insert-only input too. An updating input remains upsert-shaped under either option value.

The writer does not order two entries for the same row key, including entries in concurrent requests, so a streaming aggregation that emits repeated updates for one key has no latest-input-value guarantee. Use this bounded form only when the aggregate emits at most one final mutation per key for the entire write, not merely per window, or separate dependent writes so one completes before the next begins.

Keeping a plain insert portable#

The table-local insert-only mode narrows an input containing inserts alone to an insert-only planner contract:

CREATE TABLE profile_events (
  rowkey STRING,
  profile ROW<name STRING, email STRING>,
  PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'profiles',
  'sink.insert-only-input-mode' = 'insert-only'
);

INSERT INTO profile_events
VALUES ('user#1001', ROW('Alice', 'alice@example.com'));

This mode keeps the clause-less statement portable across Flink 1.20, 2.2, and 2.3. It does not make Bigtable insert-if-absent because a physical write still overwrites cells under an existing row key. The option does not narrow an updating query, and the planner continues to request upserts and deletes for one.

Writing a stable cell timestamp#

Writable timestamp metadata applies one TIMESTAMP_LTZ(6) value to every cell the row writes:

CREATE TABLE profile_versions (
  rowkey STRING,
  profile ROW<name STRING>,
  cell_timestamp TIMESTAMP_LTZ(6) METADATA FROM 'timestamp',
  PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'profiles',
  'sink.insert-only-input-mode' = 'insert-only'
);

INSERT INTO profile_versions
VALUES (
  'user#1001',
  ROW('Alice'),
  CAST('2026-08-30 10:15:30.123000' AS TIMESTAMP_LTZ(6))
);

The explicit timestamp makes a replay address the same cell version after Flink serializes the record again. The value above already has millisecond granularity, which Bigtable accepts without enabling sink.cell-timestamp.truncate-to-millis. A missing or null metadata value leaves the client to use the TaskManager’s wall clock instead.

Lookup joins#

Joining a Bigtable lookup table#

A processing-time temporal join turns each input row into a Bigtable point read. This example reuses the profiles table above and adds a generated facts table with a processing-time column:

CREATE TABLE events (
  event_id STRING,
  user_id STRING,
  proc_time AS PROCTIME()
) WITH (
  'connector' = 'datagen',
  'number-of-rows' = '100'
);

SELECT e.event_id, e.user_id, p.profile.name
FROM events AS e
LEFT JOIN profiles FOR SYSTEM_TIME AS OF e.proc_time AS p
  ON e.user_id = p.rowkey;

The equality condition must cover the single Bigtable row-key column. Bigtable has one atomic row key, so the connector rejects composite lookup keys and nested family fields as lookup keys. The default is a synchronous point read without a cache.

Enriching Pub/Sub events before creating tasks#

This pipeline treats Bigtable as a low-latency attribute store keyed by the user ID. It consumes JSON events from Pub/Sub, reads the matching profile family, and creates a Cloud Tasks request for an external API:

SET 'execution.checkpointing.interval' = '10 s';

CREATE TABLE incoming_events (
  event_id STRING,
  user_id STRING,
  event_type STRING,
  message_id STRING METADATA FROM 'message-id' VIRTUAL,
  proc_time AS PROCTIME()
) WITH (
  'connector' = 'pubsub',
  'project' = 'my-project',
  'subscription' = 'events-sub',
  'format' = 'json'
);

CREATE TABLE user_attributes (
  rowkey STRING,
  profile ROW<tier STRING, api_path STRING>,
  PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'user-attributes',
  'scan.row-prefix' = 'user#',
  'lookup.async' = 'true',
  'lookup.cache' = 'PARTIAL',
  'lookup.partial-cache.max-rows' = '10000',
  'lookup.partial-cache.expire-after-write' = '10 min'
);

CREATE TABLE api_tasks (
  event_id STRING,
  user_id STRING,
  event_type STRING,
  user_tier STRING,
  api_path STRING,
  source_message_id STRING,
  request_headers MAP<STRING, STRING> METADATA FROM 'headers'
) WITH (
  'connector' = 'cloud-tasks',
  'project' = 'my-project',
  'location' = 'asia-northeast1',
  'queue' = 'events',
  'http.url' = 'https://api.example.com/events',
  'http.method' = 'POST',
  'http.headers.Content-Type' = 'application/json',
  'format' = 'json'
);

INSERT INTO api_tasks
SELECT
  e.event_id,
  e.user_id,
  e.event_type,
  a.profile.tier,
  a.profile.api_path,
  e.message_id,
  MAP['X-Source-Message-Id', e.message_id]
FROM incoming_events AS e
JOIN user_attributes FOR SYSTEM_TIME AS OF e.proc_time AS a
  ON e.user_id = a.rowkey;

The inner join creates no task when the lookup finds no row or when scan.row-prefix excludes the user ID. The selected profile fields make that family the projection; an unselected family does not leave Bigtable. The example uses asynchronous PARTIAL caching, so cache misses start non-blocking point reads and results are retained under the configured row and expiry bounds.

The lookup choices change this pipeline at distinct boundaries:

The Table connector lookup section owns the complete cache and retry behavior. Pub/Sub source acknowledgements and Cloud Tasks creation remain independently at-least-once; the lookup join does not strengthen either endpoint’s delivery guarantee.

Change Streams#

Before these examples run, enable Change Streams on the physical profiles table and create the single-cluster routing application profile named single-cluster-profile. The SQL DDL registers a Flink table but does not provision either prerequisite.

Consuming Change Streams with SQL#

Envelope mode emits one insert row for every Bigtable mutation. The physical columns carry the row key and ordered mutation entries, while virtual metadata columns expose mutation identity and service timestamps:

CREATE TABLE profile_mutations (
  row_key BYTES,
  entries ARRAY<ROW<
    entry_index INT,
    kind STRING,
    family STRING,
    qualifier ROW<value_type STRING, bytes_value BYTES, long_value BIGINT>,
    `timestamp` ROW<value_type STRING, bytes_value BYTES, long_value BIGINT>,
    `value` ROW<value_type STRING, bytes_value BYTES, long_value BIGINT>,
    delete_range ROW<
      start_bound STRING,
      start_micros BIGINT,
      end_bound STRING,
      end_micros BIGINT
    >
  >>,
  mutation_type STRING NOT NULL
    METADATA FROM 'mutation-type' VIRTUAL,
  commit_timestamp TIMESTAMP_LTZ(9) NOT NULL
    METADATA FROM 'commit-timestamp' VIRTUAL,
  source_cluster_id STRING
    METADATA FROM 'source-cluster-id' VIRTUAL
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'profiles',
  'scan.mode' = 'change-stream',
  'scan.change-stream.changelog-mode' = 'envelope',
  'scan.app-profile-id' = 'single-cluster-profile'
);

SELECT
  row_key,
  mutation_type,
  commit_timestamp,
  entry_index,
  kind,
  family,
  qualifier,
  entry_timestamp,
  entry_value,
  delete_range
FROM profile_mutations
CROSS JOIN UNNEST(entries) AS entry_table(
  entry_index,
  kind,
  family,
  qualifier,
  entry_timestamp,
  entry_value,
  delete_range
);

The entries array preserves the service order, and entry_index preserves that position after UNNEST. The envelope is a mutation record rather than a reconstructed current row, so deletes also arrive as inserted envelope rows. It declares no primary key and cannot feed a keyed upsert sink without a stateful transformation that owns row reconstruction.

Producing a selected cell with keep-latest#

Store the complete non-key part of each logical profile as a JSON object in state:current. The producer’s state family has one string qualifier, current; the consumer’s flat name and tier columns are decoded from its JSON contents. Writing name and tier as separate qualifiers would require a different row-reconstruction protocol.

Create the ordinary state family without a garbage-collection rule and enable Change Streams before writing any example data. The selected-cell source rejects GC mutations affecting its cell, and SQL table auto-creation requires a GC rule, so provision this table separately. Supply the Flink JSON format jar as described in Table setup. Use the same single-cluster application profile for both jobs, and replace cluster-a below with that profile’s actual cluster ID.

Choose an epoch-millisecond startup timestamp after Change Streams was enabled and before the first write. Replace the example timestamp in the consumer DDL with that value, still within the table’s retained history. A fresh timestamp-based consumer can then read these writes even if it starts after the producer finishes; it does not take a snapshot of older rows.

Run this producer statement and wait for its job to finish:

CREATE TABLE profile_writes (
  profile_id STRING,
  state ROW<`current` STRING>,
  PRIMARY KEY (profile_id) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'profiles',
  'sink.app-profile-id' = 'single-cluster-profile',
  'sink.write-mode' = 'keep-latest',
  'sink.insert-only-input-mode' = 'insert-only'
);

INSERT INTO profile_writes
VALUES (
  'profile#1',
  ROW(JSON_OBJECT('name' VALUE 'Alice', 'tier' VALUE 'gold' NULL ON NULL))
);

keep-latest deletes all versions of state:current and sets the JSON replacement in the same row entry. Run the next statement only after the first job has completed, then wait for this job too:

INSERT INTO profile_writes
VALUES (
  'profile#1',
  ROW(JSON_OBJECT(
    'name' VALUE CAST(NULL AS STRING),
    'tier' VALUE 'silver' NULL ON NULL
  ))
);

NULL ON NULL retains the name field with a JSON null value. The JSON object still contains every non-key field, so the source can decode the complete replacement without consulting an earlier row. Submitting this replacement again writes another mutation and can produce another identical UPDATE_AFTER. Separate same-row entries remain unordered; these completed jobs demonstrate sequential application, not event-time arbitration or exactly-once effects.

Register and query the consumer in another SQL session:

CREATE TABLE profile_changes (
  profile_id STRING NOT NULL,
  name STRING,
  tier STRING,
  PRIMARY KEY (profile_id) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'profiles',
  'scan.mode' = 'change-stream',
  'scan.change-stream.changelog-mode' = 'selected-cell',
  'scan.app-profile-id' = 'single-cluster-profile',
  'scan.change-stream.selected-cell.family' = 'state',
  -- Base64 for the qualifier "current".
  'scan.change-stream.selected-cell.qualifier-base64' = 'Y3VycmVudA==',
  'scan.change-stream.selected-cell.source-cluster-id' = 'cluster-a',
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1788652800000',
  'value.format' = 'json',
  'value.json.fail-on-missing-field' = 'true',
  'value.json.ignore-parse-errors' = 'false'
);

SELECT profile_id, name, tier FROM profile_changes;

The source emits keyed UPDATE_AFTER values ('profile#1', 'Alice', 'gold') and ('profile#1', NULL, 'silver') for the two writes. SQL result collection can normalize that changelog into an initial insert, before/after updates and deletes carrying the previous value, and omit identical replacements. Interpret the displayed operations according to the SQL client’s result mode. The JSON format rejects missing fields and malformed values instead of silently dropping an incomplete replacement. The source keeps consuming until the job is cancelled or a bounded end timestamp is configured.

To remove the logical value, the producer protocol requires a full selected-column or selected-family delete without a following set, which the source emits as a key-only DELETE. A whole-cell SQL NULL is not that operation: the sink writes its null-string encoding and still emits delete-then-set. Use JSON null fields inside a complete object for nullable values. A null family skips that family; a row with no non-null family fails serialization. sink.insert-only-input-mode = insert-only applies only when a statement’s input contains inserts alone; it does not forbid updating queries. If an upstream changelog supplies a DELETE, this keep-latest sink deletes the entire Bigtable row, including unrelated cells. See the selected-cell protocol for the accepted mutation forms and failure conditions.

Replicating a selected cell into BigQuery#

Selected-cell mode interprets one configured cell as the complete logical value for its row key. The producer must atomically delete the complete selected column or family before setting the replacement cell; arbitrary Bigtable writers do not automatically satisfy that protocol.

The source declares exactly one primary key and emits keyed UPDATE_AFTER or DELETE rows. That changelog can feed a BigQuery table with the same primary key and CDC enabled:

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

CREATE TABLE current_profiles (
  profile_id STRING NOT NULL,
  name STRING,
  tier STRING,
  PRIMARY KEY (profile_id) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'profiles',
  'scan.mode' = 'change-stream',
  'scan.change-stream.changelog-mode' = 'selected-cell',
  'scan.app-profile-id' = 'single-cluster-profile',
  'scan.change-stream.selected-cell.family' = 'state',
  -- Base64 for the qualifier "current".
  'scan.change-stream.selected-cell.qualifier-base64' = 'Y3VycmVudA==',
  'scan.change-stream.selected-cell.source-cluster-id' = 'cluster-a',
  'value.format' = 'json'
);

CREATE TABLE analytics_profiles (
  profile_id STRING NOT NULL,
  name STRING,
  tier STRING,
  PRIMARY KEY (profile_id) NOT ENFORCED
) WITH (
  'connector' = 'bigquery',
  'project' = 'my-project',
  'dataset' = 'analytics',
  'table' = 'current_profiles',
  'sink.cdc.enabled' = 'true',
  'sink.create-disposition' = 'create-if-needed',
  'sink.cdc.max-staleness' = '10 min'
);

INSERT INTO analytics_profiles
SELECT profile_id, name, tier FROM current_profiles;

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 example selects one source cluster so multi-cluster conflict resolution cannot reorder the selected-cell changelog. It deliberately supplies no BigQuery sequence metadata. BigQuery therefore resolves colliding mutations for one primary key by arrival order, and Change Streams partitions do not supply a total application order across the pipeline. This is an analytics-replica pattern, not a strict replica guarantee. Because Change Streams startup defaults to latest, a fresh job materializes only mutations that arrive after it starts. A populated source needs a separate initial snapshot or backfill coordinated with a Change Streams handoff before the BigQuery table contains every existing key.

Local development#

Running against the emulator#

Google’s Bigtable emulator ships with the Cloud SDK, and the sink reaches it over a plaintext channel with no credentials:

gcloud beta emulators bigtable start --host-port=localhost:8086

# The admin surface works too, so the table can be created against the emulator.
BIGTABLE_EMULATOR_HOST=localhost:8086 \
    cbt -project my-project -instance my-instance createtable orders families=cf

BigtableSink.<OrderEvent>builder()
        .table(TableDestination.of("my-project", "my-instance", "orders"))
        .serializer(new OrderEventMutations())
        .emulatorEndpoint("localhost:8086")
        .build();

The source reaches it the same way:

BigtableSource.<Order>builder()
        .table(TableDestination.of("my-project", "my-instance", "orders"))
        .deserializer(new OrderRows())
        .emulatorEndpoint("localhost:8086")
        .build();

The project and instance ids are opaque path segments to the emulator; neither has to exist. It implements MutateRows, CheckAndMutateRow, ReadModifyWriteRow, ReadRows and the table admin surface, which is enough to develop against — but it validates far less than the service does, so a mutation it accepts is not evidence that Bigtable would.

One read-path difference is worth knowing while developing: the emulator models no tablets, so it offers almost no split boundaries and a job against it runs on one split whatever the parallelism. Reading in parallel is something only real Bigtable shows.