Spark 4.2 Auto CDC: SCD type 1 without writing MERGE

Auto CDC in Apache Spark 4.2 takes a feed of inserts, updates and deletes and keeps a table in step with it, without you writing a MERGE. You name the key, the column that says which change came first, and the condition that marks a delete. Spark applies each change in that order, even when a change arrives late.

In 4.2 it does one thing: a SCD type 1Slowly Changing DimensionsThe ways a warehouse handles a change to a descriptive record, such as a customer's address: type 1 overwrites it, type 2 adds a new dated version, and type 3 keeps the previous value in an extra column. table, 1 row per key in its latest state, from Python. The SQL form and SCD type 2 are marked for Spark 4.3. On Databricks both already exist.

Why not write the MERGE yourself

A MERGE that applies each batch of changes looks right until the changes arrive out of order. Logs are read in parallel, files land late, a retry sends an old batch again. The MERGE then does what it is told: the last change to arrive wins, not the last one to happen. Deletes are worse, because once a row is gone a late update to it looks like a new customer and inserts it back.

Apache's own proposal for Auto CDC says this plainly: hand-written MERGE for slowly changing dimensions is error-prone with deletes and out-of-order data. Step through 6 events and switch between the 2 ways of applying them:

Try itApply a change feed: arrival order or Auto CDC?

Events arrived

·1·2·3·4·5·6

Applied by

Arrival order (MERGE)Auto CDC, sequence_by seq

Change feed, as it arrives

#	id	op	city	seq
1	1	UPSERT	Leeds	1
2	2	UPSERT	York	2
3	1	UPSERT	Bath	5
4	1	UPSERT	Hull	3 late
5	2	DELETE		6
6	2	UPSERT	Derby	4 late

Target table: customers

Wrong

id	city
1	Hull
2	Derby

The code in Spark 4.2

Auto CDC lives in Spark Declarative Pipelines, so it runs in a pipeline with spark-pipelines run, not in a plain script. The target is a streaming table you create first, and the flow fills it:

from pyspark import pipelines as dp

@dp.temporary_view
def customer_changes():
    # one row per change: id, city, op ('UPSERT' or 'DELETE'), seq
    return spark.readStream.table("raw.customer_changes")

dp.create_streaming_table("customers")

dp.create_auto_cdc_flow(
    target="customers",
    source="customer_changes",
    keys=["id"],
    sequence_by="seq",
    apply_as_deletes="op = 'DELETE'",
    except_column_list=["op", "seq"],
    stored_as_scd_type=1,
)

What each argument does:

Column names must be plain, with no table prefix: "seq", not "changes.seq".

How it gets the order right

For every key, Spark compares each new change's sequence_by value with the one already applied. An older change is ignored. A delete leaves a tombstone behind for a while, so a late update to that key finds it and is dropped instead of bringing the row back. That is the work a correct MERGE would need a second table and a window function for.

So the one decision that matters is the sequence_by column. It must come from the source system, such as the database log position or the time the change was made, never the time your pipeline read it. If 2 changes to one key can have the same value, add a tiebreak column.

Where the change feed comes from

Spark 4.2 also added the other half. A table that keeps a change log can be read as one with spark.read.changes(), or a CHANGES clause in SQL:

spark.read.option("startingVersion", "10").option("endingVersion", "20").changes("shop.orders")

That works for Data Source V2 tables whose catalog exposes a change log. Feeds from databases usually come from a CDC toolChange Data CaptureCopying only the rows that were added, changed or deleted in a source database, read from its change log, instead of copying the whole table again. such as Debezium or a managed connector, landing as files or Kafka messages that the pipeline reads as a stream.

Spark 4.2 or Databricks

The Python call is the same on both, so a type 1 pipeline written for Spark 4.2 moves to Databricks unchanged. On Databricks the same flow in SQL reads:

CREATE OR REFRESH STREAMING TABLE customers;

CREATE FLOW customers_cdc AS AUTO CDC INTO customers
FROM stream(customer_changes)
KEYS (id)
APPLY AS DELETE WHEN op = 'DELETE'
SEQUENCE BY seq
COLUMNS * EXCEPT (op, seq)
STORED AS SCD TYPE 1;

Change the last line to STORED AS SCD TYPE 2 and the table keeps every version instead. Our data engineering course builds a whole pipeline this way in Lakeflow pipelines on Databricks, and what's new in Spark 4.2 covers the rest of the release.