CDC into the Lakehouse: Idempotent MERGE, SCD Type 2, Deletes and Out-of-Order Data

Change data capture (CDC) looks simple: read the database's change log and apply it to a table in the lakehouse. In production it is where pipelines quietly corrupt data. Events arrive twice, arrive out of order, arrive days late, and arrive as deletes for rows you never saw. A naive MERGE gives the right answer on the happy path and a subtly wrong one everywhere else. This post walks through the failure modes and the patterns we use on Delta Lake to make CDC correct, idempotent and replayable.

What does a CDC event actually contain?

Log-based CDC tools such as Debezium read the database's write-ahead log (the Postgres WAL, MySQL binlog or SQL Server transaction log) and emit one event per row change. Simplified, a Debezium event carries:

Field	Meaning
op	c create, u update, d delete, r snapshot read
before / after	Row image before and after the change (after is null for deletes)
source.lsn, source.ts_ms	Position in the source log, and commit time
ts_ms	When the connector processed the event

The most important design decision is the ordering key. Use the source log position (LSN, binlog file and position, or SCN) when you have it. A timestamp is not enough: several commits can share a millisecond, and clocks are not a total order. We carry this as a sortable _seq column through every layer.

Which failure modes must the pipeline handle?

1. Duplicates. Kafka and the connector give at-least-once delivery. Restarts replay events.
2. Several changes to one key in one micro-batch. MERGE fails if more than one source row matches a target row, and silently applying the "wrong" one is worse.
3. Out-of-order arrival. Retries, partition reassignment, re-snapshots and multi-source merges can deliver an older change after a newer one.
4. Deletes. A hard delete loses the information you need to reject a late, stale update for the same key.
5. Late data for history tables. For SCD Type 2, a late event can belong in the middle of a key's history, not at the end.
6. Schema changes in the source table.
7. Replays and backfills. Re-running yesterday's batch must not change the result.

Step 1: parse and deduplicate in bronze and silver

Bronze stores the raw events unchanged (append-only, with the Kafka metadata), so any later layer can be rebuilt. We also keep a parsed, append-only silver.customer_events table with every change; Step 4 relies on it. For applying changes, silver keeps only the latest change per key per micro-batch. The code below assumes the Debezium envelope has already been parsed from the Kafka value into a payload struct:

from pyspark.sql import DataFrame, Window, functions as F

def latest_per_key(batch: DataFrame) -> DataFrame:
    parsed = (
        batch
        .select(
            F.col("payload.op").alias("op"),
            F.coalesce("payload.after", "payload.before").alias("row"),
            F.col("payload.source.lsn").cast("long").alias("_seq"),
            F.timestamp_millis(F.col("payload.source.ts_ms")).alias("_source_ts"),  # source commit time
        )
        .select("op", "_seq", "_source_ts", "row.*")
    )
    w = Window.partitionBy("customer_id").orderBy(F.col("_seq").desc())
    return (
        parsed
        .withColumn("_rn", F.row_number().over(w))
        .filter("_rn = 1")
        .drop("_rn")
        .withColumn(
            "_row_hash",
            F.sha2(F.to_json(F.struct("name", "email", "segment", "city")), 256),
        )
    )

Two details matter here. First, the change hash is built with to_json(struct(...)), not concat_ws. concat_ws silently skips nulls, so ('a', null) and (null, 'a') would produce the same hash and a real change would be missed. Second, for deletes we take the before image, so the key is available.

Step 2: SCD Type 1 with a sequence guard and soft deletes

For "current state" tables, the guard condition s._seq > t._seq makes the MERGE order-independent. An older event arriving late can never overwrite a newer one.

Deletes are the trap. If you hard-delete the row, a late update with an older sequence finds no matching row and is inserted, resurrecting stale data. So we soft-delete: keep a tombstone row that remembers the sequence number at which the key was deleted.

MERGE INTO silver.customers AS t
USING customer_changes AS s
  ON t.customer_id = s.customer_id
WHEN MATCHED AND s._seq > t._seq THEN UPDATE SET
  t.name       = s.name,
  t.email      = s.email,
  t.segment    = s.segment,
  t.city       = s.city,
  t._seq       = s._seq,
  t._row_hash  = s._row_hash,
  t.is_deleted = (s.op = 'd'),
  t._updated_at = current_timestamp()
WHEN NOT MATCHED THEN INSERT
  (customer_id, name, email, segment, city, _seq, _row_hash, is_deleted, _updated_at)
  VALUES (s.customer_id, s.name, s.email, s.segment, s.city, s._seq, s._row_hash,
          s.op = 'd', current_timestamp());

Consumers read through a view that filters is_deleted = false. A periodic job physically removes tombstones older than the maximum lateness you accept, for example 30 days. This statement is also idempotent: replaying the same batch matches every row with s._seq = t._seq, so the guard is false and nothing changes.

Step 3: SCD Type 2 with a single MERGE

A Type 2 dimension keeps every version of a row, with valid_from, valid_to and is_current. Applying a change means two writes: close the current version and insert the new one. Delta's MERGE can only apply one action per source row, so we feed it each changed key twice:

MERGE INTO gold.dim_customer AS t
USING (
  -- 1) Every change, keyed: closes the current version (or inserts brand-new keys)
  SELECT s.customer_id AS merge_key, s.* FROM customer_changes s
  UNION ALL
  -- 2) Changed existing keys again, with a NULL key: never matches, so it inserts the new version
  SELECT NULL AS merge_key, s.*
  FROM customer_changes s
  JOIN gold.dim_customer t
    ON s.customer_id = t.customer_id AND t.is_current
  WHERE s.op <> 'd' AND s._row_hash <> t._row_hash
) AS u
  ON t.customer_id = u.merge_key AND t.is_current
WHEN MATCHED AND (u.op = 'd' OR t._row_hash <> u._row_hash) THEN UPDATE SET
  t.is_current = false,
  t.valid_to   = u._source_ts
WHEN NOT MATCHED AND u.op <> 'd' THEN INSERT
  (customer_id, name, email, segment, city, _row_hash, _seq, valid_from, valid_to, is_current)
  VALUES (u.customer_id, u.name, u.email, u.segment, u.city, u._row_hash, u._seq,
          u._source_ts, NULL, true);

Walk through the cases:

Case	Branch 1 (keyed)	Branch 2 (NULL key)	Result
New key	No current match: insert	Join finds nothing	One new current row
Attribute change	Matches, hash differs: close	Present: insert	Old version closed, new version current
No real change	Matches, hash equal: no action	Filtered out	Untouched (no spurious versions)
Delete	Matches, op = 'd': close	Filtered out	History ends, no current row

The pattern assumes one change per key per batch (Step 1 guarantees it) and in-order arrival per key. Comparing hashes stops no-op updates from creating junk versions, which is a common cause of bloated dimensions.

Note the trade-off in collapsing to the latest change per batch: if a key changes A → B → C inside one micro-batch, version B never reaches the dimension, so history has micro-batch granularity. When every intermediate version matters, send those keys through the Step 4 rebuild, or use AUTO CDC, which keeps intermediate versions.

Step 4: when late data lands in the middle of history

If an event arrives with a _seq lower than the highest _seq already applied for its key, appending it as the "current" version is wrong. It belongs somewhere in the middle of the key's history, and it may even predate a delete. The robust fix is to rebuild the full history of only the affected keys from the append-only event log, and replace it atomically. Rebuilding from the events, rather than patching the dimension, keeps deletes and delete-then-recreate gaps correct, because the delete events themselves are in the log.

from pyspark.sql import Window, functions as F

attrs = ["name", "email", "segment", "city"]
keys = late_changes.select("customer_id").distinct()

# Every change ever seen for the affected keys (the late ones are already appended to the log)
events = (spark.table("silver.customer_events")
            .join(keys, "customer_id")
            .select("customer_id", *attrs, "_row_hash", "_seq", "_source_ts", "op")
            .dropDuplicates(["customer_id", "_seq"]))

w = Window.partitionBy("customer_id").orderBy("_seq")      # order by log position, never by timestamp
rebuilt = (
    events
    .withColumn("prev_hash", F.lag("_row_hash").over(w))
    .withColumn("prev_op", F.lag("op").over(w))
    # keep real changes, deletes, and the first version after a delete; drop no-op updates
    .filter("prev_hash IS NULL OR op = 'd' OR prev_op = 'd' OR _row_hash <> prev_hash")
    .withColumn("valid_to", F.lead("_source_ts").over(w))   # the next change (or delete) closes a version
    .filter("op <> 'd'")                                    # deletes only close; they are not versions
    .withColumn("is_current", F.col("valid_to").isNull())
    .withColumnRenamed("_source_ts", "valid_from")
    .withColumn("_delete", F.lit(False))
    .drop("prev_hash", "prev_op", "op")
)

# Versions in the dimension that the rebuilt history no longer contains must be removed
dim = spark.table("gold.dim_customer")
stale = (dim.join(keys, "customer_id")
            .join(rebuilt.select("customer_id", "_seq"), ["customer_id", "_seq"], "left_anti")
            .select("customer_id", "_seq")
            .withColumn("_delete", F.lit(True)))

rebuilt.unionByName(stale, allowMissingColumns=True).createOrReplaceTempView("rebuilt_history")

Each version is identified by the _seq of the change that created it. The Step 3 MERGE stores that value too, so it is a stable, unique version key, unlike valid_from, because several commits can share a timestamp.

MERGE INTO gold.dim_customer AS t
USING rebuilt_history AS s
  ON t.customer_id = s.customer_id AND t._seq = s._seq
WHEN MATCHED AND s._delete THEN DELETE
WHEN MATCHED THEN UPDATE SET
  t.valid_from = s.valid_from, t.valid_to = s.valid_to, t.is_current = s.is_current,
  t.name = s.name, t.email = s.email, t.segment = s.segment, t.city = s.city,
  t._row_hash = s._row_hash
WHEN NOT MATCHED AND NOT s._delete THEN INSERT
  (customer_id, name, email, segment, city, _row_hash, _seq, valid_from, valid_to, is_current)
  VALUES (s.customer_id, s.name, s.email, s.segment, s.city, s._row_hash, s._seq,
          s.valid_from, s.valid_to, s.is_current);

Because it is one MERGE, readers never see a half-rebuilt key. Delta's transaction isolation guarantees the old or the new history, never a mix. We route a key here only when its incoming _seq is lower than the highest _seq already applied for it, so the common, in-order case stays on the cheaper Step 3 path.

Can the platform do this for us?

On Databricks, Lakeflow Declarative Pipelines (formerly Delta Live Tables) provide a declarative CDC API, AUTO CDC (previously APPLY CHANGES). It handles sequencing, out-of-order events, deletes and SCD Type 1 or 2 for you:

from pyspark import pipelines as dp          # older code: `import dlt` and dlt.create_auto_cdc_flow
from pyspark.sql.functions import col, expr

dp.create_streaming_table("dim_customer")

dp.create_auto_cdc_flow(
    target="dim_customer",
    source="customer_changes_clean",
    keys=["customer_id"],
    sequence_by=col("_seq"),                 # ordering key: source log position
    apply_as_deletes=expr("op = 'd'"),
    except_column_list=["op", "_seq", "_row_hash"],
    stored_as_scd_type=2,
)

We use it when the pipeline lives in Lakeflow, and the hand-written patterns above when it does not, or when we need custom behaviour. Either way, understanding the mechanics is what lets you verify the result and debug it at 2 a.m.

Step 5: make the stream idempotent end to end

We run silver and gold as Structured Streaming jobs with foreachBatch, usually with trigger(availableNow=True) on a schedule. That gives incremental processing with batch-style cost control.

def upsert_customers(batch_df, batch_id: int) -> None:
    changes = latest_per_key(batch_df)
    changes.createOrReplaceTempView("customer_changes")
    batch_df.sparkSession.sql(SCD1_MERGE_SQL)   # the sequence-guarded MERGE from Step 2

(spark.readStream.table("bronze.customers_cdc")
    .writeStream
    .foreachBatch(upsert_customers)
    .option("checkpointLocation", "/Volumes/prod/checkpoints/silver_customers")
    .trigger(availableNow=True)
    .start())

If a batch fails after the MERGE commits but before the checkpoint records it, Spark re-runs that batch. The sequence guard makes the re-run a no-op. For append-only writes inside foreachBatch, use Delta's idempotent-write options so a replayed batch is skipped:

(batch_df.write.format("delta")
    .option("txnAppId", "silver_orders_writer")   # stable id per writing query
    .option("txnVersion", batch_id)               # Delta ignores versions already committed
    .mode("append")
    .saveAsTable("silver.orders"))

Schema evolution, backfills and downstream propagation

• Schema evolution. New nullable source columns can flow through with Delta schema evolution. On Databricks SQL that is MERGE WITH SCHEMA EVOLUTION INTO .... We still alert on every schema change and require review for type changes and renames, which are breaking.
• Backfills and re-snapshots. Debezium snapshot reads (op = 'r') carry a log position too, so the same sequence guards apply. A full re-snapshot becomes a safe, idempotent replay rather than a special case.
• Propagating changes downstream. Enable the Change Data Feed (delta.enableChangeDataFeed = true) on silver tables. Gold jobs then read only the rows that changed (readChangeFeed), instead of rescanning whole tables.

Testing CDC logic

CDC bugs hide in orderings that rarely happen, so we test orderings explicitly:

• Unit-test the merge logic with small DataFrames covering duplicate, out-of-order, delete-then-late-update, delete-then-reinsert and no-op-update sequences.
• Property test: apply the same set of events in many random orders and assert that the final table is identical every time.
• Reconcile counts and checksums between the source database and the silver table daily, per key range, and alert on drift.

Key takeaways

• Order by the source log position, not timestamps, and carry it through every layer.
• Deduplicate to one change per key per batch before every MERGE.
• Sequence-guarded merges make out-of-order and replayed events harmless.
• Soft-delete with tombstones, or late stale updates will resurrect deleted rows.
• SCD Type 2 fits in one atomic MERGE with the union trick. Late data needs a targeted history rebuild.
• Make streams idempotent (guards, or txnAppId / txnVersion) so every batch can safely be replayed.