Back to archive

By Ivan Vydrin13 min read4 August 2026< 50 views

Designing an Idempotent CDC Pipeline with Lakeflow

CDC pipelines rarely throw. They produce a target table that is quietly wrong. What AUTO CDC handles for you, and the four decisions it still leaves open.

What AUTO CDC handles for you, and the four decisions it still expects you to make

Production Databricks Notes: architecture decisions, investigations, implementation patterns, and lessons for data and AI systems that must operate beyond the demo.

Change data capture pipelines fail in a characteristic way. They do not throw. They produce a target table that is subtly, silently wrong: a customer with two current addresses, a record whose status went backwards, a deletion that was applied and then undone by a replay.

Lakeflow pipelines, formerly Delta Live Tables, removes most of the mechanical difficulty here through its AUTO CDC APIs. They compute SCD Type 1 and Type 2 from a change feed, handle out-of-order arrival, and manage the merge logic you would otherwise write and maintain yourself.

What they do not do is make the four decisions underneath. This article covers what the API guarantees, what it assumes, and where the remaining correctness burden sits.

Naming, first

Two renames sit on top of each other here, and searching for the wrong one is how people conclude their documentation is out of date.

The product. Delta Live Tables (2022) became part of the Lakeflow family announced at DAIS 2024, and at DAIS 2025 Databricks donated the engine to Apache Spark and took Lakeflow generally available. The docs now carry an explicit naming note: the Databricks product is Lakeflow pipelines, and the open-source framework it extends is Apache Spark Declarative Pipelines (SDP). They are interoperable and differ in features, and the difference that matters here is stated plainly: the AUTO CDC APIs are available only in Lakeflow pipelines. The family also contains Lakeflow Connect (managed ingestion connectors) and Lakeflow Jobs (orchestration, formerly Databricks Workflows).

Old import dlt and @dlt decorators still work, and the event log is read the same way in both (spark.read.table("event_log")), but current docs only teach from pyspark import pipelines as dp. Treat dlt as legacy-but-functional rather than something to write new code against. No hard deprecation date has been published.

The API. The AUTO CDC APIs replace the APPLY CHANGES APIs. Databricks documents them as having the same syntax, APPLY CHANGES remains available, and Databricks recommends using AUTO CDC in its place.

Old	Current
`APPLY CHANGES INTO`	`AUTO CDC INTO`
`apply_changes()`	`create_auto_cdc_flow()`
`apply_changes_from_snapshot()`	`create_auto_cdc_from_snapshot_flow()`

The Python functions have the same signature as the ones they replace, so migration is a rename. You will still find apply_changes in most tutorials and Stack Overflow answers.

Two prerequisites worth knowing before you design around this: the pipeline must run on serverless Lakeflow pipelines or the Pro or Advanced editions, and AUTO CDC FROM SNAPSHOT is available in the Python interface only.

What idempotence means here

Idempotence in a CDC pipeline is a property of the target table, not of the job.

The correct statement: after processing a set of change events, the target reflects the source state, and processing that same set again produces no further change. Reprocessing is a no-op.

That is a stronger claim than "the job can be rerun without erroring", and it is the claim that matters, because the situations that produce duplicates are the ordinary ones. An upstream connector retries after a network blip. A replay is triggered to backfill a gap and overlaps existing data. Two events for the same key arrive in the wrong order because they took different paths.

flowchart LR
    subgraph FEED["Change feed, in arrival order"]
        direction TB
        E1["userId 7 / seq 12<br/>status = active"]
        E2["userId 7 / seq 10<br/>status = pending<br/>(arrives late)"]
        E3["userId 7 / seq 12<br/>status = active<br/>(duplicate)"]
    end

    FEED --> CDC{"AUTO CDC<br/>KEYS (userId)<br/>SEQUENCE BY seq"}

    CDC --> TGT["Target row<br/>userId 7, status = active<br/>at seq 12"]

    CDC -.->|"seq 10 loses to seq 12"| N1["Out-of-order:<br/>resolved by sequence,<br/>not arrival"]
    CDC -.->|"same key, same seq"| N2["Duplicate:<br/>no further change"]

Alt text: A change feed arrives with three events for userId 7: sequence 12 setting status to active, then a late-arriving sequence 10 setting status to pending, then a duplicate of sequence 12. AUTO CDC, keyed on userId and sequenced by seq, produces a single target row at status active, sequence 12. Sequence 10 loses to sequence 12 because ordering follows the sequence column rather than arrival, and the duplicate produces no further change.

AUTO CDC gives you this property for the merge itself. It does not give it to you for everything upstream and downstream of the merge, which is where the four decisions live.

The first decision: what defines a record's identity

KEYS in SQL, keys in Python. The columns that identify a logical record in the source.

This looks obvious and frequently is not. The failure is choosing a key that is unique in the source system but not unique in the change feed, or choosing a surrogate key that the source regenerates.

The test I use: if the source system deleted this record and recreated it with the same business meaning, would the key be the same? If yes, you have chosen a business key, which is usually what you want. If no, a recreation will appear as a new record and the history of the old one will end abruptly with no explanation.

Composite keys are supported and are frequently correct. A multi-tenant source almost always needs the tenant identifier in the key, and omitting it is a data leak that presents as a data quality problem.

The second decision: what defines order

SEQUENCE BY in SQL, sequence_by in Python. The column specifying the logical order of change events, which the pipeline uses to handle events that arrive out of order.

This is the parameter that decides correctness, and it is the one most often filled in with whatever timestamp was nearest to hand.

What the API does with it is the useful part here. Given a key and a sequence value, AUTO CDC applies the highest-sequence event and discards lower ones, whatever order they arrived in. That is the whole out-of-order guarantee, and it is entirely a function of this column being right.

Two constraints bound what you can put in it: the column must be a sortable data type, and NULL sequencing values are not supported. Enforce the second one upstream with an expectation rather than discovering it during an incident.

Ties are resolved with a struct, ordering by the first field and falling through to the second. The API handles that mechanically; deciding which columns go in the struct is where the judgment lives.

Which is the companion note's subject, and it is a longer conversation than it looks, because every candidate column encodes a different claim about what "later" means. ingested_at says arrival order is truth. A source log sequence number says the source's commit order is truth. An effective_date says business chronology is truth. Those produce different tables from identical input, and the note works through what each one costs.

The third decision: what a delete means

APPLY AS DELETE WHEN operation = "DELETE"
APPLY AS TRUNCATE WHEN operation = "TRUNCATE"

# arguments to create_auto_cdc_flow(...)
apply_as_deletes = expr("operation = 'DELETE'"),
apply_as_truncates = expr("operation = 'TRUNCATE'"),

Without these clauses, a delete event is just another row and lands in the target as a record with operation = 'DELETE' sitting in it. This is a real failure mode and it is quiet, because the pipeline is green and the row count goes up rather than down.

A deleted row is temporarily retained as a tombstone in the underlying Delta table, and a view in the metastore filters tombstones out. Query the view, not the underlying table, unless you want to see them. This surprises people who check physical storage to verify a deletion and conclude it did not work.

The retention interval defaults to two days and is configurable through the pipelines.cdc.tombstoneGCThresholdInSeconds table property. That number matters more than it looks: a delete whose corresponding late data arrives after the tombstone is collected will not be suppressed. If your source can deliver a record hours or days behind its own delete event, raise the threshold past your worst observed lateness.

APPLY AS TRUNCATE is the clause with an SCD-type restriction, not deletes: apply_as_truncates is supported for SCD type 1 only. It is also a much larger operation than a delete, and wiring it to a condition that can fire unexpectedly is one of the few destructive things you can configure here.

The fourth decision: Type 1 or Type 2

STORED AS SCD TYPE 1 overwrites. STORED AS SCD TYPE 2 retains history. Both are generally available. A BITEMPORAL option exists for tracking changes across both business and system time, and it is Beta at the time of writing, so treat it as a direction rather than something to build a compliance story on.

Which is better depends on one question: will anyone ever need to know what this record looked like at a past point in time? For a customer address feeding a shipping system, probably not. For a customer address feeding a tax calculation or a regulatory report, certainly, and discovering that after eighteen months of Type 1 means the history does not exist to recover.

Type 2 has a schema requirement that is easy to miss: when you specify the target table schema explicitly, you must include the __START_AT and __END_AT columns with the same data type as the sequence_by field. That coupling is the useful detail. It means your choice of sequence column also determines the type of your validity interval, and it is another reason to make that choice deliberately.

TRACK HISTORY ON narrows what generates a new version. Without it, every change to any column closes the current row and opens a new one, including changes to columns nobody analyses. On a wide dimension with a chatty source, this is the difference between a history table you can query and one that grows without producing information.

STORED AS SCD TYPE 2
TRACK HISTORY ON * EXCEPT (last_synced_at, etl_batch_id)

A complete flow

Python:

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

@dp.temporary_view
def users():
    return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_current")

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

SQL:

CREATE OR REFRESH STREAMING TABLE users_current;

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

Note except_column_list / COLUMNS * EXCEPT. The CDC control columns should not land in the target. Leaving them in is harmless until somebody joins on the target and picks up sequenceNum as if it meant something in the business domain.

The full SQL grammar, for reference:

CREATE OR REFRESH STREAMING TABLE table_name;
CREATE FLOW flow_name AS AUTO CDC [ONCE] INTO table_name
FROM source
KEYS (keys)
[IGNORE NULL UPDATES [ON {columnList | * EXCEPT (exceptColumnList)}]]
[APPLY AS DELETE WHEN condition]
[APPLY AS TRUNCATE WHEN condition]
SEQUENCE BY orderByColumn
[SYSTEM SEQUENCE BY systemOrderByColumn]
[COLUMNS {columnList | * EXCEPT (exceptColumnList)}]
[STORED AS {SCD TYPE 1 | SCD TYPE 2 | BITEMPORAL}]
[TRACK HISTORY ON {columnList | * EXCEPT (exceptColumnList)}]
[COLUMNS TO UPDATE columnName]

`IGNORE NULL UPDATES`, and why it is a data contract question

ignore_null_updates controls how nulls in incoming updates are handled, and defaults to False.

A null in a change event has two readings. Either the source is telling you this field is now empty, or the source is sending a partial update and null means "unchanged". Both are common. Which one applies is a property of your upstream connector, and it is not discoverable from the data.

Getting it wrong in one direction silently overwrites good values with nulls. Getting it wrong in the other direction makes it impossible to ever clear a field. Both present as a data quality complaint months later.

Write down which one your source does. That sentence belongs in the reliability contract, and it is one of the twelve questions from Week 1.

What is still yours

AUTO CDC handles ordering and merge semantics. Four things remain outside it.

Source-side duplicates. If the connector emits the same change event twice with the same sequence value, the merge is idempotent with respect to it, which is good. If it emits the same logical change twice with different sequence values, they are two distinct events by the API's definition and Type 2 history will show two versions. Whether that is wrong depends on your contract.

Checkpoint state. Week 1 covered why the checkpoint is production state rather than a cache. The CDC-specific consequence: clearing it replays the whole change feed into a target whose merge is idempotent per event but whose Type 2 history is not. You get the correct current row and a history table with duplicate validity intervals, which is the worst available outcome because it looks fine from the serving query.

Downstream consumers. Idempotent CDC into a target says nothing about what happens when a downstream job reads that target twice. Each hop needs its own answer.

Custom foreachBatch sinks. If you write outside the declarative flow, none of these guarantees follow you. The documentation is blunt about it: "foreachBatch() provides only at-least-once write guarantees."

For a Delta target you get exactly-once back by keying the write on the batch ID, which Delta then deduplicates for you:

app_id = "orders-streaming-job"
main_table = "main.orders.orders_current"

def process_orders(batch_df, batch_id):
    batch_df.write \
        .format("delta") \
        .mode("append") \
        .option("txnVersion", batch_id) \
        .option("txnAppId", app_id) \
        .saveAsTable(main_table)

Delta checks the (txnAppId, txnVersion) pair, and "if Spark replays a batch with the same batchId, Delta Lake skips the duplicate write."

Two things to get right. txnAppId must be stable across restarts, because a generated one makes every restart look like a new application and defeats the deduplication entirely. And for a non-Delta sink there is no equivalent, so the idempotence has to be implemented in the target system.

A test worth having

The one integration test I would write first: process a set of change events, capture the target table, process the identical set again, assert the target is unchanged.

That single test catches the majority of the failures above. It fails if the key is wrong, if the sequence column ties, if deletes are not wired, and if a foreachBatch sink is non-idempotent. It is also the test that almost never exists, because the happy path already passed and nobody thought to run it twice.

Add a second: process the events in a shuffled order and assert the same final state. That one verifies your sequencing choice rather than your assumption about arrival order.

Verified against Databricks documentation on 2026-07-29: The AUTO CDC APIs, AUTO CDC INTO, create_auto_cdc_flow. Check the current reference before relying on any specific behaviour; this API has been renamed once already.