Spanner examples#

Worked cases beyond the quickstart follow the shared source-to-sink order. The Spanner options page lists every option, while the Spanner connector page explains the runtime contracts behind them.

DataStream source#

The quickstart owns the basic bounded DataStream read. The cases below change its read shape, snapshot, compute placement, or row handling.

Reading a key range instead of a query#

A table read takes a key set and a column list, and is the cheapest shape when the rows wanted are a contiguous range of the primary key. There is no SQL to be root-partitionable, so nothing about the read can be refused for being undistributable.

SpannerSource.<Order>builder()
        .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
        .readOperation(
                SpannerReadOperation.read(
                        "Orders",
                        KeySet.range(
                                KeyRange.closedOpen(
                                        Key.of("order#1000"), Key.of("order#2000"))),
                        Arrays.asList("OrderId", "Total")))
        .deserializer(new OrderDeserializer())
        .build();

SpannerReadOperation.readUsingIndex("Orders", "OrdersByCustomer", keys, columns) reads the same way through a secondary index, with the key set interpreted in the index’s key space.

Reading a large table without disturbing serving traffic#

Data Boost serves the read from compute that is not the instance’s. The caller needs spanner.databases.useDataBoost on the database, the read is billed separately, and its concurrency has a quota of its own.

SpannerSource.<Order>builder()
        .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
        .readOperation(
                SpannerReadOperation.query(
                        Statement.of(
                                "SELECT OrderId, Total FROM Orders WHERE Total > 100")))
        .deserializer(new OrderDeserializer())
        .dataBoostEnabled(true)
        // Both are hints. Asking for one partition per subtask is reasonable; getting a
        // different number is normal, and the enumerator warns when the plan is smaller
        // than the parallelism.
        .maxPartitions(env.getParallelism())
        .build();

Reading at a fixed timestamp#

Reading two tables at the same timestamp makes their contents consistent with each other, which a job joining them usually wants.

Timestamp readAt = Timestamp.now();

SpannerSource.<Order>builder()
        .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
        .readOperation(
                SpannerReadOperation.query(Statement.of("SELECT OrderId FROM Orders")))
        .deserializer(new OrderDeserializer())
        .timestampBound(TimestampBound.ofReadTimestamp(readAt))
        .build();

The timestamp has to lie inside the database’s version_retention_period — an hour by default, up to a week — or the data behind it is gone. TimestampBound.ofExactStaleness(...) is the other accepted form; ofMaxStaleness and ofMinReadTimestamp are rejected, because Spanner allows them only on a single-use transaction.

Skipping rows on the way in#

Emitting nothing from a source deserializer skips the row. recordsSkipped counts each such input row once.

new SpannerStructDeserializationSchema<Order>() {
    @Override
    public void deserialize(Struct row, Collector<Order> out) {
        // Rows the query could not exclude, filtered before they cost anything downstream.
        if (!row.isNull("Total")) {
            out.collect(new Order(row.getString("OrderId"), row.getLong("Total")));
        }
    }

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

A row you could not read is a different thing: throw, and the job fails rather than losing it. Collector calls must be synchronous, must emit non-null records, and must not continue after deserialize returns.

DataStream sink#

The quickstart owns the basic DataStream write. The cases below change mutation routing, mutation shape, refusal handling, or batching.

Writing to several tables from one sink#

The dynamic destinations guide explains why Spanner takes its table from each mutation instead of using the resolver-based pattern. The mutation names its table, so routing by record needs no option, only a serializer that decides.

SpannerSink.<Event>builder()
        .database(DatabaseDestination.of("my-project", "my-instance", "events-db"))
        .serializer(
                (event, context) ->
                        Mutation.newInsertOrUpdateBuilder(
                                        event.isAudit() ? "AuditEvents" : "Events")
                                .set("EventId")
                                .to(event.getId())
                                .set("Body")
                                .to(event.getBody())
                                .build())
        .build();

At start-up, the sink reads index-aware cell weights for the database’s visible tables, so both pre-existing tables in this example are weighed correctly against the batch limit. A table created after the writer opens, or hidden from the writer’s database role, is counted without its index entries. Both tables must already exist because the sink does not create them.

Deleting rows#

A delete is a mutation like any other, so a stream of keys is a stream of deletes.

SpannerSink.<String>builder()
        .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
        .serializer((orderId, context) -> Mutation.delete("Orders", Key.of(orderId)))
        .build();

Deletes are idempotent, so replaying a record is harmless. A delete over a key range is also possible, but the sink counts it as one row against maxBatchCells because the client cannot know how many rows the range matches.

Skipping records#

Returning null skips the record: it is written nowhere, is not a failure, and increments recordsSkipped.

SpannerSink.<Event>builder()
        .database(DatabaseDestination.of("my-project", "my-instance", "events-db"))
        .serializer(
                (event, context) ->
                        event.isHeartbeat()
                                ? null
                                : Mutation.newInsertOrUpdateBuilder("Events")
                                        .set("EventId")
                                        .to(event.getId())
                                        .build())
        .build();

Dropping refused mutations instead of failing the job#

By default the first refused mutation fails the job. failedMutationHandler changes that decision.

SpannerSink.<String>builder()
        .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
        .serializer(
                (orderId, context) ->
                        Mutation.newInsertOrUpdateBuilder("Orders")
                                .set("OrderId")
                                .to(orderId)
                                .set("Total")
                                .to(0L)
                                .build())
        .failedMutationHandler(FailureHandler.logAndDrop())
        .build();

Read what reaches this handler before relying on it. By default only ALREADY_EXISTS and INVALID_ARGUMENT do; a NOT NULL violation, an over-long value, a foreign key, or a CHECK constraint fails the job instead. The constraint-violation policy can route those schema refusals to the same handler:

SpannerSink.<String>builder()
        .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
        .serializer(
                (orderId, context) ->
                        Mutation.newInsertOrUpdateBuilder("Orders")
                                .set("OrderId")
                                .to(orderId)
                                .set("Total")
                                .to(0L)
                                .build())
        .constraintViolationPolicy(ConstraintViolationPolicy.ROUTE_TO_FAILURE_HANDLER)
        .failedMutationHandler(FailureHandler.logAndDrop())
        .build();

Choose that policy when the stream genuinely carries occasional records the schema will not accept. Keep the fail-job default when a refusal would mean the column mapping is wrong. The error table lists every case and the reason for its route.

A handler using the connector’s failure type can inspect the mutation, table, and error message. This example logs the table and service-provided text without capturing a non-serializable logger:

SpannerSink.<String>builder()
        .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
        .serializer(
                (orderId, context) ->
                        Mutation.newInsertOrUpdateBuilder("Orders")
                                .set("OrderId")
                                .to(orderId)
                                .set("Total")
                                .to(0L)
                                .build())
        .failedMutationHandler(
                (FailureHandler<FailedMutation>)
                        failure ->
                                System.getLogger("SpannerFailures")
                                        .log(
                                                System.Logger.Level.WARNING,
                                                "Dropping a mutation on {0}: {1}",
                                                failure.getTable(),
                                                failure.getErrorMessage()))
        .build();

Treat that text as potentially sensitive because the service may include rejected identifiers or values. Omit or sanitize it when the log is not an approved destination for record data.

Tuning the batch#

The defaults are Apache Beam’s and sit far below the request limits. Lower them for latency, and raise them only after checking what maxBatchCells counts. The configuration reference records each ceiling and its evidence. The limits are combined, so a batch flushes when the first one binds and raising one limit alone often changes nothing. Setting maxBatchMutations above maxBatchCells writes a warning where the job’s main runs because the cell cap must bind first. batchWriteTimeout separately bounds one complete write attempt so a stalled response stream cannot hold the task thread indefinitely.

SpannerSink.<String>builder()
        .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
        .serializer(
                (orderId, context) ->
                        Mutation.newInsertOrUpdateBuilder("Orders")
                                .set("OrderId")
                                .to(orderId)
                                .set("Total")
                                .to(0L)
                                .build())
        .writerOptions(
                SpannerWriterOptions.builder()
                        .maxBatchMutations(100)
                        // Bound one complete BatchWrite attempt, including a response
                        // stream that returns some groups and then stalls.
                        .batchWriteTimeout(Duration.ofSeconds(20))
                        // A commit delay trades latency for throughput by letting Spanner
                        // group this commit with others. Zero to 500 ms.
                        .maxCommitDelay(Duration.ofMillis(50))
                        // A backfill that must not disturb serving traffic on the same
                        // instance.
                        .rpcPriority(SpannerRpcPriority.LOW)
                        .build())
        .build();

maxBatchCells counts every written table cell plus the secondary-index entries that the write changes. Read bufferedCells beside bufferedBytes to identify the limit that is firing.

Table source#

Reading a bounded table with SQL#

The default Table source reads one bounded snapshot through the same partitioned source as the DataStream API. Create and seed the physical table in the quickstart’s orders-db database first:

CREATE TABLE inventory (
  sku STRING(32) NOT NULL,
  quantity INT64 NOT NULL,
  updated_at TIMESTAMP NOT NULL
) PRIMARY KEY (sku);

INSERT INTO inventory (sku, quantity, updated_at)
VALUES ('widget-1', 12, TIMESTAMP '2026-08-30T00:00:00Z');

The Flink DDL defaults to scan.mode = 'bounded', so no Change Stream options are needed:

CREATE TABLE inventory (
  sku STRING,
  quantity BIGINT,
  updated_at TIMESTAMP_LTZ(9),
  PRIMARY KEY (sku) NOT ENFORCED
) WITH (
  'connector' = 'spanner',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'database' = 'orders-db',
  'table' = 'inventory'
);

SELECT sku, quantity
FROM inventory
WHERE quantity > 0;

Projection is pushed into the Spanner column list. The non-key quantity predicate remains with Flink because bounded scan pushdown is limited to consecutive primary-key columns.

Table sink#

Writing upserts with SQL#

A declared primary key makes the Spanner sink use insertOrUpdate for inserts and updates. This bounded aggregation emits one final row for each composite key. Create its physical destination in orders-db first:

CREATE TABLE account_status (
  region STRING(16) NOT NULL,
  account INT64 NOT NULL,
  status STRING(64)
) PRIMARY KEY (region, account);

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

CREATE TABLE status_events (
  region STRING,
  account BIGINT,
  status STRING
) WITH (
  'connector' = 'datagen',
  'number-of-rows' = '1000',
  'fields.region.length' = '16',
  'fields.status.length' = '64'
);

CREATE TABLE account_status (
  region STRING,
  account BIGINT,
  status STRING,
  PRIMARY KEY (region, account) NOT ENFORCED
) WITH (
  'connector' = 'spanner',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'database' = 'orders-db',
  'table' = 'account_status'
);

INSERT INTO account_status
SELECT region, account, MAX(status)
FROM status_events
GROUP BY region, account;

Without the Flink PRIMARY KEY declaration, the connector accepts only insert-only input and uses Spanner insert. The declaration is a planner contract; it does not create or verify the physical key.

Lookup joins#

Joining a composite-key lookup table#

A processing-time temporal join turns each facts row into a Spanner point read. Create and seed the physical table in the quickstart’s orders-db database first:

CREATE TABLE accounts (
  region STRING(16) NOT NULL,
  account INT64 NOT NULL,
  name STRING(128)
) PRIMARY KEY (region, account);

INSERT INTO accounts (region, account, name)
VALUES ('region-1', 1, 'Ada');

The one-row data generator derives region-1 from account 1, so it deterministically produces the seeded key. Real event tables normally carry both key columns as physical fields.

CREATE TABLE account_events (
  account BIGINT,
  region AS CONCAT('region-', CAST(account AS STRING)),
  proc_time AS PROCTIME()
) WITH (
  'connector' = 'datagen',
  'number-of-rows' = '1',
  'fields.account.kind' = 'sequence',
  'fields.account.start' = '1',
  'fields.account.end' = '1'
);

CREATE TABLE accounts (
  region STRING,
  account BIGINT,
  name STRING,
  PRIMARY KEY (region, account) NOT ENFORCED
) WITH (
  'connector' = 'spanner',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'database' = 'orders-db',
  'table' = 'accounts'
);

SELECT e.region, e.account, a.name
FROM account_events AS e
LEFT JOIN accounts FOR SYSTEM_TIME AS OF e.proc_time AS a
  ON e.account = a.account AND e.region = a.region;

The equality condition must cover every declared primary-key column. The join predicates may appear in either order, but the connector encodes the lookup key in the PRIMARY KEY (region, account) declaration order. That order must match the physical Spanner primary key, or the point read addresses a different key.

Change Streams#

Comparing changelog modes and materializing changes#

The Table source offers two changelog shapes over the same physical Change Stream. Create the watched table, a stream that captures old and new values, and a separate materialization table first:

CREATE TABLE source_orders (
  order_id INT64 NOT NULL,
  customer STRING(128),
  status STRING(32)
) PRIMARY KEY (order_id);

CREATE CHANGE STREAM source_order_changes FOR source_orders
OPTIONS (value_capture_type = 'NEW_ROW_AND_OLD_VALUES');

CREATE TABLE order_replica (
  order_id INT64 NOT NULL,
  customer STRING(128),
  status STRING(32),
  source_commit_timestamp TIMESTAMP NOT NULL,
  server_transaction_id STRING(MAX) NOT NULL,
  record_sequence STRING(MAX) NOT NULL,
  mod_number INT64 NOT NULL,
  source_table STRING(MAX) NOT NULL,
  mod_type STRING(16) NOT NULL
) PRIMARY KEY (order_id);

Mode	Required capture	Flink primary key	Row kinds	Delete row
`full`	`NEW_ROW_AND_OLD_VALUES`	Optional	`INSERT`, `UPDATE_BEFORE`, `UPDATE_AFTER`, `DELETE`	Complete old row
`upsert`	`NEW_ROW` or `NEW_ROW_AND_OLD_VALUES`	Required and equal to the record key	`INSERT`, `UPDATE_AFTER`, `DELETE`	Key only

Both modes expose the same readable commit, transaction, sequence, mod, table, and modification-type metadata. mod_number starts at zero for each original data-change record, and the before and after rows of one full-mode update share it. Both DDLs start at latest, so a fresh job waits for changes committed after it starts.

The full DDL needs no primary key because it reconstructs complete retract rows:

CREATE TABLE full_order_changes (
  order_id BIGINT,
  customer STRING,
  status STRING,
  source_commit_timestamp TIMESTAMP_LTZ(9)
    METADATA FROM 'commit-timestamp' VIRTUAL,
  server_transaction_id STRING METADATA FROM 'server-transaction-id' VIRTUAL,
  record_sequence STRING METADATA FROM 'sequence' VIRTUAL,
  mod_number INT METADATA FROM 'mod-number' VIRTUAL,
  source_table STRING METADATA FROM 'table' VIRTUAL,
  mod_type STRING METADATA FROM 'mod-type' VIRTUAL
) WITH (
  'connector' = 'spanner',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'database' = 'orders-db',
  'table' = 'source_orders',
  'scan.mode' = 'change-stream',
  'scan.change-stream.name' = 'source_order_changes',
  'scan.change-stream.changelog-mode' = 'full',
  'scan.startup.mode' = 'latest'
);

SELECT source_commit_timestamp, server_transaction_id, record_sequence,
       mod_number, source_table, mod_type, order_id, customer, status
FROM full_order_changes;

The sink’s advertised upsert changelog excludes UPDATE_BEFORE, and the planner does not send those rows to it. Use upsert when the source should feed a keyed Spanner sink:

CREATE TABLE order_changes (
  order_id BIGINT,
  customer STRING,
  status STRING,
  source_commit_timestamp TIMESTAMP_LTZ(9)
    METADATA FROM 'commit-timestamp' VIRTUAL,
  server_transaction_id STRING METADATA FROM 'server-transaction-id' VIRTUAL,
  record_sequence STRING METADATA FROM 'sequence' VIRTUAL,
  mod_number INT METADATA FROM 'mod-number' VIRTUAL,
  source_table STRING METADATA FROM 'table' VIRTUAL,
  mod_type STRING METADATA FROM 'mod-type' VIRTUAL,
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'spanner',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'database' = 'orders-db',
  'table' = 'source_orders',
  'scan.mode' = 'change-stream',
  'scan.change-stream.name' = 'source_order_changes',
  'scan.change-stream.changelog-mode' = 'upsert',
  'scan.startup.mode' = 'latest'
);

CREATE TABLE order_replica (
  order_id BIGINT,
  customer STRING,
  status STRING,
  source_commit_timestamp TIMESTAMP_LTZ(9),
  server_transaction_id STRING,
  record_sequence STRING,
  mod_number BIGINT,
  source_table STRING,
  mod_type STRING,
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'spanner',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'database' = 'orders-db',
  'table' = 'order_replica'
);

INSERT INTO order_replica
SELECT order_id, customer, status, source_commit_timestamp,
       server_transaction_id, record_sequence, CAST(mod_number AS BIGINT),
       source_table, mod_type
FROM order_changes;

After starting either job, wait until changeStreamQueriesStarted is non-zero for at least one source reader subtask, then insert a watched row from another session:

INSERT INTO source_orders (order_id, customer, status)
VALUES (1, 'Ada', 'PENDING');

The Change Streams DDL and sink DDL remain separate because scan.mode = 'change-stream' makes a table source-only. The source metadata columns are virtual, while the replica stores their selected values in ordinary physical columns for observability. The query widens the Flink INT mod number to BIGINT, the lossless mapping for a physical Spanner INT64 column. A key-only delete still works because the sink constructs a delete mutation from the primary key and does not write non-key fields.

This is a replica-shaped materialization pattern, not a strict replica guarantee. The sink is at-least-once, and BatchWrite does not guarantee the application order of successive writes to one key, so the destination is not guaranteed to retain the latest source value.

Use TIMESTAMP_LTZ(3) plus WATERMARK FOR source_commit_timestamp AS SOURCE_WATERMARK() when event-time operations matter more than retaining nanoseconds.

Filtering Change Streams records#

Table and column filters run after the connector decodes Spanner’s record and before the user deserializer runs. Each Java regular expression matches a complete table name or table.column identifier.

SpannerChangeStreamSource<OrderChange> source =
        SpannerChangeStreamSource.<OrderChange>builder()
                .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
                .changeStreamName("all_changes")
                .deserializer(new OrderChangeDeserializer())
                .startPosition(StartPosition.latest())
                .tableIncludeList(List.of("orders", "order_items"))
                .columnExcludeList(
                        List.of("orders\\.internal_note", "order_items\\.debug_payload"))
                .skipMessagesWithoutChange(false)
                .build();

Primary keys remain in keys and columnTypes even when a column expression matches them. The default above delivers a record whose projected old and new values are empty. Set skipMessagesWithoutChange(true) only when downstream processing does not need that transaction activity.

These filters do not change the Change Stream query or prevent excluded values from entering the source process. Restrict the Change Stream’s DDL watch definition when exclusion must happen inside Spanner.

Local development#

Running against the emulator#

docker run -p 9010:9010 -p 9020:9020 gcr.io/cloud-spanner-emulator/emulator:1.5.56

The emulator has separate resources from the real service. Point a dedicated gcloud configuration at its REST endpoint, then recreate the quickstart resources there:

gcloud config configurations create spanner-emulator --no-activate
gcloud config set auth/disable_credentials true --configuration=spanner-emulator
gcloud config set project my-project --configuration=spanner-emulator
gcloud config set api_endpoint_overrides/spanner http://localhost:9020/ \
    --configuration=spanner-emulator
gcloud spanner instances create my-instance --config=emulator-config \
    --description="my-instance" --nodes=1 --configuration=spanner-emulator
gcloud spanner databases create orders-db --instance=my-instance \
    --ddl='CREATE TABLE Orders (OrderId STRING(64) NOT NULL, Total INT64) PRIMARY KEY (OrderId)' \
    --configuration=spanner-emulator

SpannerSink.<String>builder()
        .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
        .serializer(
                (orderId, context) ->
                        Mutation.newInsertOrUpdateBuilder("Orders")
                                .set("OrderId")
                                .to(orderId)
                                .set("Total")
                                .to(0L)
                                .build())
        .emulatorEndpoint("localhost:9010")
        .build();

The source takes the same option:

SpannerSource.<Order>builder()
        .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
        .readOperation(
                SpannerReadOperation.query(Statement.of("SELECT OrderId FROM Orders")))
        .deserializer(new OrderDeserializer())
        .emulatorEndpoint("localhost:9010")
        .build();

Setting the endpoint also stops the client looking for credentials. Pin an image at v1.5.31 or newer: the emulator implements the BatchWrite RPC this sink writes with only from that release, and an older one answers UNIMPLEMENTED to everything. The emulator has no IAM and serializes concurrent transactions, so it is a convenience for fast feedback rather than evidence about the service.

On the read path in particular, the emulator plans exactly two partitions whatever the data, ignores both partition hints, and applies a partitionability check of its own that refuses query shapes the real service accepts — the deviation table has the details.