Change data capture with AUTO CDC

What it is

AUTO CDC is the API in Lakeflow pipelines (see Lakeflow pipelines) that takes a stream of change records and keeps a target streaming table in sync with them, as either SCD Type 1 (current state only) or SCD Type 2 (full history). You declare the target table, the key columns, and the column that orders the events; the pipeline works out the inserts, updates and deletes, including what to do when events arrive out of order.

AUTO CDC FROM SNAPSHOT solves the same problem when the source emits no change feed at all, only periodic full dumps. It compares each snapshot with the previous one, derives the change feed itself, and then applies it the same way.

Why it exists

Applying a change feed by hand means a MERGE per micro-batch, which means a staging table, a window function to pick the last change per key, and a set of assumptions about ordering that nobody writes down. See Upsert with MERGE INTO for what that looks like when you do it yourself. It works, and it is the right tool for a one-off, but as a pattern it is copied from pipeline to pipeline and gets subtly wrong every time: a late-arriving update overwrites a newer value, a delete is applied before the insert it supersedes, a full refresh reprocesses history in a different order and lands somewhere else.

SCD Type 2 is worse. Closing the previous version of a row, opening a new one, and keeping the validity intervals consistent when an event turns up two hours late is genuinely difficult logic, and it is the same logic in every warehouse in the world. AUTO CDC makes it a declaration: keys, sequencing column, SCD type.

How it works

Requirements

The CDC APIs need the pipeline to run on serverless compute, or on the Pro or Advanced editions of Lakeflow pipelines. They are not part of open-source Apache Spark Declarative Pipelines.

Declaring a flow

You create the target streaming table first, then a flow that writes into it:

CREATE OR REFRESH STREAMING TABLE users_current;

CREATE FLOW apply_cdc AS AUTO CDC INTO users_current
FROM stream(main.bronze.users_cdf)
KEYS (user_id)
APPLY AS DELETE WHEN operation = "DELETE"
SEQUENCE BY sequence_num
COLUMNS * EXCEPT (operation, sequence_num)
STORED AS SCD TYPE 1;

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

dp.create_streaming_table("users_current")

dp.create_auto_cdc_flow(
    target="users_current",
    source="users",
    keys=["user_id"],
    sequence_by=col("sequence_num"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "sequence_num"],
    stored_as_scd_type=1,
)

The default behaviour for INSERT and UPDATE events is an upsert on the keys. STORED AS defaults to SCD Type 1 if you leave it out.

The clauses that matter

Sequencing and out-of-order events

SEQUENCE BY is the whole trick. AUTO CDC processes events in the order that column defines, not the order they arrive, so an update stamped 5 that turns up after an update stamped 6 is discarded rather than applied on top. The column must be a sortable type, must be monotonically increasing in the sense that matters (one distinct update per key per value), and nulls are not supported. To break ties on a timestamp, sequence by a STRUCT of two columns: SEQUENCE BY STRUCT(event_ts, event_id) orders by the first field and falls back to the second.

For SCD Type 2 sources, a deleted row is kept briefly as a tombstone in the underlying Delta table so that a late event for that key can still be ordered correctly, with a view in the metastore filtering the tombstones out. The retention window is the pipelines.cdc.tombstoneGCThresholdInSeconds table property.

__START_AT and __END_AT

An SCD Type 2 target gains two generated columns holding the validity interval of each version, taken from the values of the sequencing column rather than from wall-clock time. A row whose __END_AT is NULL is the current version. If you declare the target table’s schema explicitly, you must include both columns with the same type as the sequencing column.

By default any change to any column opens a new version. TRACK HISTORY ON * EXCEPT (city) narrows that: changes to city update the current row in place, changes to anything else create a version.

AUTO CDC FROM SNAPSHOT

Available in the Python interface only. Instead of a change feed you give it a snapshot, and it diffs consecutive snapshots to derive inserts, updates and deletes. Two patterns:

• one snapshot per pipeline run, versioned by the run itself, when snapshots arrive regularly and in order;
• a version function, which you write to return the next (DataFrame, version) pair, when several snapshots are waiting or ordering needs to be explicit. Snapshots are processed in ascending version order; one that turns up out of order is skipped, and returning None means there is nothing new.

Snapshots can come from a Delta table, from files in cloud storage, or over JDBC.

Bitemporal tracking

STORED AS BITEMPORAL with SYSTEM SEQUENCE BY extends SCD Type 2 across two time dimensions: business time and system time, so you can ask both “what was true then” and “what did we know then”. It is in Beta, so treat it as something to be aware of rather than something to design around.

Example: SCD Type 2 with history on a subset of columns

CREATE OR REFRESH STREAMING TABLE main.silver.customers_history;

CREATE FLOW customers_cdc AS AUTO CDC INTO main.silver.customers_history
FROM stream(main.bronze.customers_cdf)
KEYS (customer_id)
APPLY AS DELETE WHEN operation = "DELETE"
SEQUENCE BY STRUCT(op_ts, op_id)
COLUMNS * EXCEPT (operation, op_ts, op_id)
STORED AS SCD TYPE 2
TRACK HISTORY ON * EXCEPT (last_seen_at);

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr, struct

@dp.view
def customers():
    return spark.readStream.table("main.bronze.customers_cdf")

dp.create_streaming_table("main.silver.customers_history")

dp.create_auto_cdc_flow(
    target="main.silver.customers_history",
    source="customers",
    keys=["customer_id"],
    sequence_by=struct("op_ts", "op_id"),        # tie-break on op_id
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "op_ts", "op_id"],
    stored_as_scd_type="2",
    track_history_except_column_list=["last_seen_at"],
)

A customer who moves twice ends up with three rows: two closed intervals and one with __END_AT IS NULL. A change to last_seen_at alone updates the current row and creates nothing.

Common mistakes

• Sequencing by ingestion time instead of source event time. Two events that hit the bronze table in the wrong order then get applied in the wrong order. Use the sequence number or commit timestamp the source system emits.
• A nullable sequencing column. Nulls are not supported, and the failure is not obvious from the pipeline UI. Enforce it with an expectation, see Data quality: expectations and constraints.
• Expecting APPLY AS TRUNCATE WHEN to work on SCD Type 2. It is supported for Type 1 only, because truncating a history table has no sensible meaning.
• Streaming from an AUTO CDC target as if it were an ordinary table. The target is rewritten in place by the flow, so a downstream streaming read has to go through its change data feed, not a plain STREAM read.
• Declaring the target schema for SCD Type 2 and omitting __START_AT and __END_AT. They have to be there, with the same data type as the sequencing column.
• Reaching for AUTO CDC on a source that emits full snapshots. That is what AUTO CDC FROM SNAPSHOT is for, and it is Python-only.