"Changes, Not Snapshots" — CDC into the Lakehouse

Skim Card

One thing to remember: a change stream is not a table; replicating OLTP into a lakehouse means folding every insert, update and delete into table state, in order, exactly once in effect.

• Keep two tables per source table: an append-only history of every change, and a current-state replica built by MERGE.
• Order changes per key by the source’s log position (LSN), never by updated_at or arrival time.
• Hard deletes must arrive as delete events; the poller never saw 1,176 UK Q3 orders go, worth exactly £57,064.96.
• Micro-batches replay after a crash: skip a batch already committed, and MERGE only changes newer than the row (higher LSN; streamed beats snapshot).
• Every streaming MERGE adds delete files each read must apply; schedule compaction from day one (eleven batches: 17 files, 26,600 deletes).
• Reconcile the replica against the source row by row: 2.26 million rows, zero differences.

Use …	When …
Spark Structured Streaming + MERGE in foreachBatch	You run Spark already, minutes of lag are fine, and you want history and state from one job
Flink’s Iceberg upsert sink	Seconds of lag matter and you budget for frequent compaction of equality deletes
Iceberg Kafka Connect sink	Append-only topics (Part 27); in Iceberg 1.12 it doesn’t apply updates or deletes
A periodic full extract	You can’t read the source’s log (Part 15)

Verified against: PostgreSQL 18.6, Debezium 3.7.0.Final (Kafka Connect 4.3.1), Apache Kafka 4.4.0, Spark 4.1.3 + spark-sql-kafka-0-10 4.1.3, Apache Iceberg 1.12.0, Apache Polaris 1.8.0, DuckDB 2.0.0 · vendor docs · 28 October 2026

Jump to: Concepts · Hands-On · Landscape · Design Drill

Story Opening

Ingrid had answered Rohan’s question from Part 15: no full nightly extract of the order database, re-reading every order to find a few thousand deletes. Larkspur owns that database, so read its log.

“Debezium puts every change on a Kafka topic,” Rohan said. “We append the topic to an Iceberg table and query it. Done by lunch.”

Ingrid opened the topic. “This order appears twice: once placed, once deleted. Which row does Farah’s revenue query sum? A topic of changes is a ledger of what happened. A table is what’s true now. Your job is the fold between them.”

The Quick Map

Concepts, with use cases

The Log Never Lies (Part 1) covers CDC itself (the WAL, slots, publications, Debezium’s envelope); this part starts in Kafka.

Replication is a fold, not a copy

A change stream is the ordered list of inserts (c), updates (u), deletes (d) and snapshot reads (r) a database made. A replica is the table you get by applying them in order: upsert the row for each key, remove it on a delete. You know this as event sourcing’s projection, except the events are row images, so the projection is mechanical.

What breaks without it. Query the appended stream and every order exists once per change; a cancelled one never disappears. Larkspur: shop.orders, payments and refunds become raw.orders, payments and refunds. Elsewhere: a bank replicating account balances into a risk warehouse.

Ordering: the log position, per key

Debezium stamps every event with its LSN (log sequence number), the change’s position in PostgreSQL’s write-ahead log. Two changes to one row can’t interleave (the row lock serialises them), so per key a higher LSN is a later change; Kafka keeps per-key order because Debezium keys messages by primary key (both guarantees). One exception: every snapshot row carries the one position Debezium read before snapshotting, and a transaction in flight at that moment can stream changes with lower LSNs, so a streamed change always outranks a snapshot read.

What breaks without it. Order by updated_at and Part 15’s slow clocks reorder updates; by arrival, and a re-sent event overwrites a newer one. Larkspur: step 4. Elsewhere: ride-hailing trip status, where a late en route must never overwrite completed.

Initial snapshot, then stream

A slot only holds changes made after it was created, so the connector first reads every existing row (op = 'r'), then streams from the slot with no gap (the hand-off). For a lakehouse, the history starts at the snapshot: rows deleted before it never appear in any event.

What breaks without it. Stream without a snapshot and the replica holds only rows changed since; snapshot without a slot and you lose changes made while copying. Larkspur: the snapshot carries 1 July, the stream Q3 (step 2). Elsewhere: a telecom backfilling five years of subscribers before moving billing analytics to CDC.

History table and current-state table

The history (raw.orders_changes) is append-only: every event, with operation, LSN, transaction ID, commit time and Kafka position. The current state (raw.orders) is the fold: one row per live key, merged every batch.

What breaks with only one. State alone can’t say what an order looked like on 2 August or how many were cancelled; history alone makes every query rebuild the fold, though it can rebuild state after a bug. Larkspur: the incident query reads both (step 7). Elsewhere: an insurer auditing every policy change beside the in-force book.

Hard deletes and soft deletes

A hard delete removes the row; a soft delete keeps it and marks it (status = 'voided', a deleted_at column). To CDC a soft delete is an update; a hard delete is a d event with a null after, which the replica must apply and the history should keep.

What breaks without it. A poller can’t see a row that stopped existing. Larkspur: a cancelled order is hard-deleted while its payment is voided (soft): the ledger keeps 6,332 voided payments whose orders are gone (step 6). Elsewhere: a no-show report counting appointments the clinic hard-deleted.

MERGE vs equality deletes

A MERGE (Spark, Trino) joins the batch with the table, finds each matched row’s file and position, and writes position deletes (format v2; deletion vectors in v3), or rewrites the file under copy-on-write, plus the new rows. An equality delete (Flink’s upsert mode) writes “delete every row where order_id = 42” without reading the table. The spec allows them in v2 and v3, not in v4 (still in development).

What breaks if you pick blind. Equality deletes make writes cheap and reads expensive (every reader checks rows against them until compaction); MERGE pays at write time, reading the key column. Larkspur: minutes of lag suffice, so Spark MERGEs. Elsewhere: a fraud team needing seconds takes Flink’s equality deletes and compacts hourly.

Three sinks, compared

Larkspur runs the third row on Spark it already operates. Elsewhere: a media company sinking playback events through Connect and account tables through Flink.

Under the Hood

One micro-batch, and where it can fail

For each micro-batch, Spark writes the batch’s Kafka offset ranges to offsets/N in its checkpoint, runs your function, then writes commits/N. Crash between your Iceberg commits and commits/N, and the batch reruns with the same offsets and ID: Spark’s guide says foreachBatch gives “only at-least-once write guarantees” unless you deduplicate by batch ID. Two guards make a replay harmless:

1. The append tags its snapshot summary with part16-cdc:<batchId> (Iceberg’s snapshot-property.* write option) and is skipped if the history table already has that tag.
2. The MERGE applies a change only if it’s newer than the row: a higher LSN, or any streamed change over a snapshot read. A replay matches rows at equal positions and changes nothing; so does a stale re-send of an older change.

One gap remains: a stale create arriving after its key’s delete is NOT MATCHED and comes back. That takes Connect re-sending a stretch of events after a crash, and the stretch also holds the delete, so it lasts at most a batch; tombstone rows behind a view close it. (Reasoning; not provoked in the lab.)

Hands-On

The code is in the every-record-counts folder of the companion-workspaces repository (tag every-record-counts/part-16). It adds a stream profile: Kafka 4.4.0 (KRaft, one node, 512 MB heap; Season 3’s version) and Kafka Connect 4.3.1 inside the Debezium 3.7.0.Final image (768 MB heap). PostgreSQL now runs with wal_level=logical (a restart; Polaris reconnected by itself).

make up-core && make up-stream && make up-spark
uv run python -m scripts.part16_cdc reset
uv run python -m scripts.part16_cdc setup         # step 1
uv run python -m scripts.part16_cdc connect       # step 2
uv run python -m scripts.part16_cdc replay        # step 1
uv run python -m scripts.part16_cdc topics
uv run python -m scripts.part16_cdc events OGB00038256   # step 3
uv run python -m scripts.part16_cdc stream --fail-after-batch 3   # steps 4 and 5
uv run python -m scripts.part16_cdc stream
uv run python -m scripts.part16_cdc reconcile     # step 6
uv run python -m scripts.part16_cdc incident      # step 7
uv run python -m scripts.part16_cdc maintain      # step 8

Step 1: a source database with a past

sql/part16/postgres/shop.sql creates the order service’s tables in schema shop, a least-privilege debezium role and a DBA-owned publication; shop.orders gets REPLICA IDENTITY FULL (what that logs).

The lab can’t wait three months, so it replays the generator’s history (seed 42): orders placed, fulfilled or cancelled (a hard DELETE), payments authorised, captured or voided, refunds. setup loads the database as of 1 July 00:00 UTC with COPY; after the snapshot, replay applies every later change in event-time order, one transaction per hour of event time, as fast as PostgreSQL takes them. At 5 October 01:00 the database equals that morning’s extract row for row (tests/test_part16.py), bar 15 orders cancelled after it ran.

shop.orders       41,745 rows as of 2026-07-01 00:00 UTC
shop.payments    477,022 rows as of 2026-07-01 00:00 UTC
shop.refunds      11,460 rows as of 2026-07-01 00:00 UTC

replayed 2026-07-01 00:00 -> 2026-10-05 01:00 UTC: 2,305 transactions, 2,016,330 changes in 56 s
  shop.orders    c=  140,453  u=  135,333  d=    4,837
  shop.payments  c=1,543,217  u=  140,170  d=        0
  shop.refunds   c=   52,320  u=        0  d=        0

Step 2: the connector and its initial snapshot

kafka/connectors/larkspur-shop.json (trimmed: tasks.max, the database.* connection keys, key.converter and key.converter.schemas.enable, and the topic-creation defaults):

{
  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  "plugin.name": "pgoutput",
  "topic.prefix": "larkspur",
  "table.include.list": "shop.orders,shop.payments,shop.refunds",
  "slot.name": "larkspur_shop",
  "publication.name": "larkspur_shop",
  "publication.autocreate.mode": "disabled",
  "snapshot.mode": "initial",
  "tombstones.on.delete": "false",
  "heartbeat.interval.ms": "10000",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false",
  "producer.override.compression.type": "zstd"
}

initial snapshot in Kafka after 7 s:
  larkspur.shop.orders        41,745 events for 41,745 rows
  larkspur.shop.payments     477,022 events for 477,022 rows
  larkspur.shop.refunds       11,460 events for 11,460 rows

After the replay, topics shows 322,368, 2,160,409 and 63,780 events: snapshot plus every change.

Step 3: one cancelled order, in the log

OGB00038256, £22.96, was placed on 1 August at 12:16 UTC, extracted as placed by the poller’s 01:00 run on 2 August, and deleted at 01:23. Its events (trimmed):

{"op": "c", "lsn": 77728934368, "txId": 30542, "snapshot": "false", "before": null,
 "after": {"order_id": "OGB00038256", "status": "placed", ...}, ..., "total_minor": 2296}
{"op": "d", "lsn": 77732007448, "txId": 30555, "snapshot": "false",
 "before": {"order_id": "OGB00038256", "status": "placed", ...}, "after": null, ..., "total_minor": 2296}

The poller saw the first and could never see the second.

Step 4: the stream job

spark/jobs/part16_cdc.py reads the topics with trigger(availableNow=True): everything Kafka holds, in batches of maxOffsetsPerTrigger = 250,000, then stop; run every few minutes, that’s minutes of lag and no idle cluster. Per batch and table it parses the envelope into the row (after, or before for a delete) plus _op, _lsn, _tx_id, _source_ts (commit time), _snapshot and the Kafka position, then (excerpt):

if committed(spark, hist, tag):            # this batch's history is already in Iceberg: a replay
    appended = "already committed, skipped"
else:
    ch.writeTo(hist).option("snapshot-property.part16.batch", tag).append()
newest = Window.partitionBy(key).orderBy(   # streamed changes before snapshot reads, then the higher LSN
    F.col("_from_snapshot"), F.col("_lsn").desc(), F.col("_kafka_offset").desc()
)
latest = (
    ch.withColumn("_from_snapshot", F.col("_op") == "r")
    .withColumn("_rn", F.row_number().over(newest))
    .where("_rn = 1")
)

MERGE INTO pipeline.raw.orders t
USING latest_orders s
ON t.order_id = s.order_id
WHEN MATCHED AND {NEWER} AND s._op = 'd' THEN DELETE
WHEN MATCHED AND {NEWER} THEN UPDATE SET t.customer_id = s.customer_id, ...
WHEN NOT MATCHED AND s._op <> 'd' THEN INSERT (order_id, ...) VALUES (s.order_id, ...)

{NEWER} is ((s._lsn > t._lsn AND s._from_snapshot = t._from_snapshot) OR (t._from_snapshot AND NOT s._from_snapshot)); equal is not newer. pipeline is a Spark catalog logged in as Part 9’s pipeline principal: Polaris vends it storage credentials per table. A not-matched delete is skipped: its key was inserted and deleted within the batch.

batch 2: 6.3 s
  orders   c=16,124 u=14,963 d=559            changes +31,646                    current overwrite +16,880 rows +1,315 position deletes
  payments r=52,843 c=145,813 u=13,433        changes +212,089                   current append +198,656 rows +0 position deletes
  ...
batch 3: 5.5 s
  orders   c=15,674 u=15,422 d=550            changes +31,646                    current overwrite +17,041 rows +1,917 position deletes
  payments c=194,189 u=17,900                 changes +212,089                   current overwrite +195,630 rows +1,441 position deletes
  refunds  c=6,260                            changes +6,260                     current append +6,260 rows +0 position deletes
RuntimeError: simulated crash after batch 3 committed, before Spark recorded it

Batches take 5–11 s in a 1.5 GB driver. In batch 2, 31,646 orders events became 16,880 rows and 1,315 deleted positions: a key’s changes inside one batch collapse into its last.

Step 5: crash, restart, replay

--fail-after-batch 3 raises after batch 3’s six Iceberg commits but before Spark writes commits/3. The next run:

batch 3: 10.3 s
  orders   c=15,674 u=15,422 d=550            changes already committed, skipped current overwrite +0 rows +0 position deletes
  payments c=194,189 u=17,900                 changes already committed, skipped current overwrite +0 rows +0 position deletes
  refunds  c=6,260                            changes already committed, skipped current overwrite +0 rows +0 position deletes
batch 4: 6.4 s
  ...
batch 10: 5.2 s
spark-submit part16_cdc.py --max-offsets 250000: exit 0, 65 s, Spark container peak 2,316 MiB

Same batch ID, same offsets: appends skipped by tag, and every MERGE matched rows at equal positions and changed nothing (Iceberg still committed an empty snapshot). Without the tag check, batch 3’s 249,995 events would be in the history twice.

Step 6: reconcile against PostgreSQL

sql/part16/reconcile.sql (DuckDB) attaches PostgreSQL read-only and compares counts, sums and every column of every row both ways (EXCEPT ALL):

┌──────────┬─────────┬───────────┬──────────────┬──────────────┬────────────────┐
│  table   │ pg_rows │ lake_rows │   pg_minor   │  lake_minor  │ rows_differing │
├──────────┼─────────┼───────────┼──────────────┼──────────────┼────────────────┤
│ orders   │  177361 │    177361 │  21716018584 │  21716018584 │              0 │
│ payments │ 2020239 │   2020239 │ 225625780353 │ 225625780353 │              0 │
│ refunds  │   63780 │     63780 │   6512442396 │   6512442396 │              0 │
└──────────┴─────────┴───────────┴──────────────┴──────────────┴────────────────┘

(Type rows and a second query trimmed.) The sums mix currencies: checksums, not revenue. Zero differing rows in 2,261,380, in 5 s. By status, payments match too, including 6,332 voided ones: the soft-deleted halves of cancelled orders.

Step 7: the hard-deleted orders, closed

sql/part16/incident.sql sums UK Q3 online orders the board deck’s way (UTC dates, report-day euro rate) from the poller’s extracts and from the replica, and counts what the history says was deleted:

source (UK online orders, Q3 by UTC date)   orders  in EUR             GBP
1 poller extracts (board deck)              56,674   3,970    2,659,886.14
2 CDC replica raw.orders                    55,498   3,898    2,602,821.18
3 deleted, per raw.orders_changes            1,993     134       95,340.40
4   of which the poller extracted            1,176      72       57,064.96
board deck minus CDC replica                                     57,064.96

£57,064.96: exactly waterfall step 2 in docs/injected-defects.md, from 1,176 orders the poller saw placed and never saw deleted (the other 817 were cancelled before any nightly run, so the deck never counted them). After Part 6’s duplicates the deck goes from £29,881,155.45 to £29,824,090.49, and Part 4’s refund dating then gives the £29,781,578.58 Part 5 started from: all five causes of the £208,558.37 gap are explained.

Step 8: what eleven MERGEs left behind

The current-state tables are merge-on-read: each MERGE writes new and changed rows plus a position-delete file per data file it touched (Spark’s delete-granularity default is file), which every reader applies on every scan until compaction. sql/part16/spark/maintain.sql counts live files, runs rewrite_data_files (delete-file-threshold 1, remove-dangling-deletes), counts again and reads the snapshot log:

  table_name  data_files  position_delete_files  position_deletes  equality_delete_files
  raw.orders          11                      9             14286                      0
raw.payments          11                      8             12314                      0
...
  raw.orders           1                      0                 0                      0
raw.payments           1                      0                 0                      0

operation  data_files_rewritten  delete_files_removed  deletes_removed
  replace                    11                     8            12750
  replace                  NULL                     1             1536

Eleven batches left 17 delete files and 26,600 deleted positions; no equality deletes, since Spark writes only position deletes. Every file was far below the 512 MiB target, so bin-packing rewrote each table’s 11 into one. For raw.orders the rewrite commit dropped 8 of 9 delete files with their data files; the ninth, left dangling, went in a second commit. Part 19 schedules this. tests/test_part16.py replays the whole sequence from a reset and asserts each number above (6 passed in 252 s). Peak: 4.56 GiB of containers on a MacBook Pro M3 Pro, 9.8 GB Docker VM.

The Landscape

As of 28 October 2026, from each vendor’s docs:

Tips & Gotchas

Gotcha — REPLICA IDENTITY DEFAULT sends less than a row. A delete carries only the key: enough for a MERGE, useless for “what was deleted, and for how much” (step 7 works because shop.orders is FULL). And an update that doesn’t touch a large (TOASTed) column sends __debezium_unavailable_value in its place (unavailable.value.placeholder), which a full-row MERGE writes over the real text. Use FULL where it matters (whole old rows in the WAL), or skip placeholder columns in the UPDATE SET.

Gotcha — commit time isn’t event time. _source_ts is when PostgreSQL committed: months after the event in the lab’s replay, and after any backlog or backfill in production. Report on the row’s business timestamps.

Gotcha — a quiet publication still pins WAL. After the replay the slot held back 63 MB of WAL written by other databases on the server (Polaris’s catalog among them). The lab caps it with max_slot_wal_keep_size; production keeps the slot moving with a heartbeat that writes (how), and PostgreSQL 18 adds idle_replication_slot_timeout for abandoned slots.

Gotcha — a new checkpoint means a new history. startingOffsets applies only to a new query; after that the checkpoint decides. Delete the checkpoint and batch IDs restart at 0 over different offset ranges, so the part16-cdc:<batchId> tags would skip new events or let old ones in twice. Rebuild the history table with it, as reset does, from a topic that still holds everything since the snapshot (seven days’ retention here).

Tip — keep snapshots longer than any replay. The batch tags live in snapshot summaries, so expire_snapshots deletes them. Keep them well beyond any window in which Spark could re-run a batch, or the check finds nothing and appends it again.

Troubleshooting

Design Drill

The question: “An airline runs 30 operational services (flight operations, crew rostering, maintenance, baggage, loyalty), each with its own PostgreSQL database: 400 tables in all. Replicate them into the lakehouse with under 15 minutes of lag for operations dashboards and analysts, and keep it correct as services change their schemas, add tables and delete records.”

Clarify first

• Which tables need 15 minutes, which could be daily, and how busy is the busiest?
• Do services hard-delete? Must deleted loyalty members disappear from the lake (erasure), or stay in history?
• Who owns schema changes, and is there any contract today, or do services migrate when they like?
• Are the databases managed (plugins, who creates slots)? Is “15 minutes” commit-to-queryable?

The shape of a strong answer

• Debezium per database (30 connectors and slots), pgoutput, DBA-owned publications; heartbeats with an action query; slot lag alerted.
• Snapshots: an initial one per connector; tables added later get Debezium incremental snapshots (an execute-snapshot signal of type incremental in the signal.data.collection table) while streaming continues.
• Schema changes: added columns evolve the Iceberg tables; breaking ones (renames, narrowing) are blocked by registry rules (Part 27) and contracts (Part 20), and a table that breaks anyway is quarantined alone.
• Per table: history partitioned by ingestion day; current state merged by key (streamed over snapshot, then LSN) and partitioned by bucket(key), so a MERGE reads only the buckets its batch touches.
• One metadata-driven Spark job, not 400, run every 5 minutes with availableNow; idempotent appends and guarded merges.
• Compaction per table by change rate; snapshots kept longer than any replay. Lag = now minus the replica’s latest _source_ts; daily reconciliation of counts and hash totals.
• Erasure: deletes flow through; erased members’ history rows go too, then old snapshots are expired.

Trade-offs to say out loud

• Spark every 5 minutes fits 15; seconds would mean Flink, equality deletes and far more compaction.
• REPLICA IDENTITY FULL makes deletes self-describing and multiplies WAL for wide, busy tables; choose it per table.
• One connector per database isolates failures but means 30 slots to watch.

Common weak answers

• “Append the topics, take the latest row per key in a view”: every query pays for the fold.
• No snapshot plan: the lake misses every row that didn’t change, and new tables need a stop-the-world backfill.
• Ignoring erasure in the history table, the snapshots and the Kafka topics.

Further Reading

1. Debezium, “Debezium connector for PostgreSQL” (3.7): snapshots, event structure, REPLICA IDENTITY, properties. https://debezium.io/documentation/reference/3.7/connectors/postgresql.html
2. Apache Spark, “Structured Streaming Programming Guide” (4.1): foreachBatch, triggers, checkpoints. https://spark.apache.org/docs/4.1.3/streaming/apis-on-dataframes-and-datasets.html
3. Apache Iceberg, table spec: row-level deletes (position, equality, deletion vectors). https://iceberg.apache.org/spec/
4. Apache Iceberg, “Flink Writes” (upsert mode and equality deletes). https://iceberg.apache.org/docs/latest/flink-writes/
5. Martin Kleppmann and Chris Riccomini, Designing Data-Intensive Applications, 2nd edition (O’Reilly, 2026), the chapter on stream processing (change data capture, log compaction).

Story Closing

Farah read the incident table twice. “So all five causes now have a number and an explanation.” Of the five promises from Part 2, this part strengthened completeness: every insert, update and delete the order database made reaches the lake, and a reconciliation proves it.

“Explained isn’t the same as built,” Ingrid said. “The star schema Farah reads is still a Python script Rohan runs by hand.”

In Part 17, Rohan rebuilds the Season 1 star schema as a dbt project on these tables: models, tests and snapshots under version control, reviewed like any other code.

This is Part 16 of “Every Record Counts: Data Engineering for Backend Engineers” — Season 2: Pipelines in Production.