Streaming Data into BigQuery: A Practical Guide

Streaming data into BigQuery is straightforward when every record is new: events, clicks, sensor readings. You append them, and the BigQuery loading guide covers how, including retries. Things change when the source is a table whose rows are updated and deleted, such as orders, customers or accounts. Appending every change gives you a history log, not the current state, and turning it back into current state with queries is slow and easy to get wrong.

BigQuery's change data capture (CDC) handles this inside the table. You stream each change with a marker saying whether it is an upsert or a delete, and BigQuery applies it to a table keyed by primary key (BigQuery: change data capture). This guide covers how it works and where it goes wrong.

How CDC streaming works

• Writes go through the Storage Write API, on the default stream, in protobuf format.
• Each row is marked as a change in the _CHANGE_TYPE pseudocolumn, which accepts only UPSERT or DELETE.
• The table must have a primary key, declared NOT ENFORCED. BigQuery does not check uniqueness, so the key must be genuinely unique in your source.
• An optional sequence number, _CHANGE_SEQUENCE_NUMBER, sets the order in which changes for the same key are applied. Without it, the most recently ingested change wins.
• A staleness setting, max_staleness on the table, tells BigQuery how out of date query results may be, which controls when changes are applied.

The table definition looks like this:

CREATE TABLE shop.orders (
  order_id INT64 PRIMARY KEY NOT ENFORCED,
  status STRING,
  updated_at TIMESTAMP
)
CLUSTER BY order_id
OPTIONS (max_staleness = INTERVAL 15 MINUTE);

Order matters: a simulation

Change streams are rarely clean. Delivery is often at least once, so duplicates arrive, and changes for the same row can arrive out of order after a retry or a partition rebalance. The code below applies one such stream three ways: last arrival wins, which is what happens without a sequence number; BigQuery's documented rule of comparing sequence numbers numerically; and a plain text comparison, which is what a hand-written deduplication query often does.

The sequence numbers are database log positions in the PostgreSQL style, two hexadecimal numbers separated by a slash, such as 16/B374D848 (PostgreSQL: pg_lsn). That shape fits BigQuery's documented format: up to four hexadecimal sections separated by /, compared numerically section by section.

# A simulation of BigQuery CDC apply rules as documented: one row per primary key, the highest
# _CHANGE_SEQUENCE_NUMBER wins, ties go to the most recently ingested row, and DELETE removes the key.

def seq_key(seq):
    # Documented format: up to four hexadecimal sections separated by "/", compared numerically section by section
    return tuple(int(part, 16) for part in seq.split("/"))

def apply(changes, order_by):
    table = {}
    for arrival, change in enumerate(changes):
        key, current = change["id"], table.get(change["id"])
        if current is None or order_by(change, arrival) >= order_by(current, current["arrival"]):
            table[key] = {**change, "arrival": arrival}
    return {k: (v["status"], v["seq"]) for k, v in sorted(table.items()) if v["type"] == "UPSERT"}

# Changes from a database log, delivered at least once and not in order. seq is the log position,
# written in hex without zero padding, as many databases print it.
changes = [
    {"id": 1, "type": "UPSERT", "status": "placed",    "seq": "F/FFFF0000"},
    {"id": 2, "type": "UPSERT", "status": "placed",    "seq": "F/FFFF0100"},
    {"id": 1, "type": "UPSERT", "status": "cancelled", "seq": "10/200"},       # later change, arrives early
    {"id": 1, "type": "UPSERT", "status": "shipped",   "seq": "F/FFFFFF00"},   # earlier change, arrives late
    {"id": 2, "type": "UPSERT", "status": "placed",    "seq": "F/FFFF0100"},   # duplicate delivery
    {"id": 3, "type": "UPSERT", "status": "placed",    "seq": "10/9A0"},
    {"id": 3, "type": "DELETE", "status": None,        "seq": "10/1A00"},      # deleted afterwards
]

print("S1 last arrival wins (no sequence number supplied)")
print("  ", apply(changes, lambda c, arrival: arrival))
print("S2 _CHANGE_SEQUENCE_NUMBER compared numerically, as BigQuery documents")
print("  ", apply(changes, lambda c, arrival: (seq_key(c["seq"]), arrival)))
print("S3 the same sequence numbers compared as plain text (a common mistake in hand-written dedup)")
print("  ", apply(changes, lambda c, arrival: (c["seq"], arrival)))

print("\nS4 why text comparison fails without zero padding")
for a, b in [("10/200", "F/FFFFFF00"), ("10/1A00", "10/9A0")]:
    print(f"   {a:>12} vs {b:<12} numeric says {'first' if seq_key(a) > seq_key(b) else 'second'} is later,",
          f"text says {'first' if a > b else 'second'}")

S1 last arrival wins (no sequence number supplied)
   {1: ('shipped', 'F/FFFFFF00'), 2: ('placed', 'F/FFFF0100')}
S2 _CHANGE_SEQUENCE_NUMBER compared numerically, as BigQuery documents
   {1: ('cancelled', '10/200'), 2: ('placed', 'F/FFFF0100')}
S3 the same sequence numbers compared as plain text (a common mistake in hand-written dedup)
   {1: ('shipped', 'F/FFFFFF00'), 2: ('placed', 'F/FFFF0100'), 3: ('placed', '10/9A0')}

S4 why text comparison fails without zero padding
         10/200 vs F/FFFFFF00   numeric says first is later, text says second
        10/1A00 vs 10/9A0       numeric says first is later, text says second

What the simulation shows

Without a sequence number, late changes win. Order 1 was shipped and later cancelled, but the "shipped" change arrived last, so last-arrival-wins left it shipped. The duplicate for order 2 did no harm, because applying the same upsert twice gives the same row.

With numeric sequence numbers, the result is right. Order 1 ended cancelled, order 2 placed once, and order 3 deleted, regardless of arrival order. Supply _CHANGE_SEQUENCE_NUMBER whenever your source has a position or version you can rely on. The documentation also says that if you use it, every write should carry it; mixing rows with and without it makes the order unpredictable.

Comparing as text is worse than not ordering at all. Log positions printed without zero padding do not sort correctly as strings: 10/200 comes before F/FFFFFF00 as text, although it is later. Text comparison kept the stale "shipped" status and brought deleted order 3 back. BigQuery's own apply avoids this, but the same mistake turns up in hand-written deduplication over a raw change log. If you order such values yourself, compare them numerically or zero-pad them.

max_staleness: freshness against cost

Changes are not applied to the table's stored data the moment they arrive. Background apply jobs run at intervals set by max_staleness. If a query asks for data while unapplied changes are still within the staleness window, it can read the last applied state. If they are older than the window, BigQuery merges them at query time, which returns current data but adds latency and cost.

The documentation's guidance is to set max_staleness to at least twice the 95th-percentile apply duration plus a buffer, and it gives a query to calculate that. A very low value means more background and runtime merges, which cost more than ordinary queries. Ask the business how fresh the data must really be: fifteen minutes is often fine for operational reporting, and much cheaper than near-zero.

Limitations to know before you design

• No DML on CDC tables: UPDATE, DELETE and MERGE statements are not supported, so corrections must also flow through the stream.
• Primary keys of up to 16 columns, and they are not enforced.
• No mixing: sending CDC rows and plain inserts on the same connection causes a validation error.
• Pseudocolumns are write-only: you cannot query _CHANGE_TYPE or _CHANGE_SEQUENCE_NUMBER from the table. If you need an audit history, write the change log to a separate append-only table too.
• Cost of background work: without a reservation, background apply jobs use on-demand pricing; for large or busy tables the documentation recommends a reservation.

Where this fits

CDC tables suit replicating operational databases for analytics with minutes of delay. For event streams where nothing is updated, plain appends are simpler and cheaper. For transformations in flight, such as enrichment, windowing or joins, a pipeline such as Pub/Sub to Dataflow to BigQuery comes first, and it can write CDC rows at the end. Clustering on the key helps queries, though the documentation notes that background and runtime merges read the full table; see partitioning and clustering in BigQuery.

Streaming pipelines and BigQuery design are part of the Data Engineering course.

Sources

• Stream table updates with change data capture (BigQuery documentation) — Google Cloud
• pg_lsn type (PostgreSQL documentation) — PostgreSQL Global Development Group