# Change data capture with Delta Lake: apply inserts, updates and deletes with MERGE

Change data capture sends every insert, update and delete a database makes, in order, with a sequence number. To apply it to a Delta table, keep the newest change per key in each batch, then MERGE: update a row only when the change's sequence number is higher than the row's, insert keys not seen before, and mark deletes with a flag rather than removing the row. A rerun, or a change that arrives late, then leaves the table as it is. Delta's change data feed gives the same kind of log for a Delta table's own changes.

- Page: https://tecyfy.com/data-engineering/course/getting-data-in/change-data-capture
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 6: Getting data in, Change data capture with MERGE
- Parts of this module: Load new files once, with a checkpoint (https://tecyfy.com/data-engineering/course/getting-data-in); Auto Loader and Lakeflow Connect (https://tecyfy.com/data-engineering/course/getting-data-in/auto-loader); Change data capture with MERGE (https://tecyfy.com/data-engineering/course/getting-data-in/change-data-capture); Build it: ingest every file once (https://tecyfy.com/data-engineering/course/getting-data-in/build-it)

At the end of this part you can:

- Apply a batch of inserts, updates and deletes to a Delta table with MERGE, safe to rerun
- Keep late or repeated changes from undoing newer ones, with a sequence number and soft deletes
- Read the changes a Delta table made, with its change data feed

[Part 1](/data-engineering/course/getting-data-in) loaded the shop's order files into bronze, every line once. Files are 1 way data arrives. The shop's orders also live in a database, and a database can send something better than files: every change it makes, as it makes it. This part applies those changes to a Delta table with the MERGE from [module 4](/data-engineering/course/delta-lake-and-iceberg/merge), this time with deletes.

## What a database sends

A database keeps a log of the changes it makes. [Change data capture](https://tecyfy.com/concepts/change-data-capture) reads that log and sends each change on: the operation, insert, update or delete, the row after it, and its position in the log, a number that only grows. Lakeflow Connect's database connectors read it from MySQL, PostgreSQL and SQL Server. Here the change log is written by hand, in batches, as a connector would deliver them.

## Apply a batch

Save this as `apply_changes.py`, in a new folder with `delta_session.py` from [module 4](/data-engineering/course/delta-lake-and-iceberg):

```python
"""Apply 1 batch of changes from the shop's order database to the orders table."""
import sys
from decimal import Decimal
from delta.tables import DeltaTable
from pyspark.sql import Window
from pyspark.sql import functions as F
from delta_session import delta_spark

# What the database's change log sends: the operation, the row after it, and the log's sequence number
BATCHES = {
    "1": [("I", 1001, "placed", Decimal("25.50"), 1), ("I", 1002, "placed", Decimal("12.00"), 2),
          ("U", 1001, "paid", Decimal("25.50"), 3), ("I", 1003, "placed", Decimal("60.00"), 4)],
    "2": [("U", 1001, "shipped", Decimal("25.50"), 5), ("D", 1002, None, None, 6),
          ("U", 1003, "paid", Decimal("60.00"), 7)],
    "late": [("U", 1001, "paid", Decimal("25.50"), 3), ("I", 1002, "placed", Decimal("12.00"), 2)],
}
CHANGES = "op STRING, order_id BIGINT, status STRING, amount DECIMAL(10,2), seq BIGINT"

spark = delta_spark("cdc")
spark.sql("""
    CREATE TABLE IF NOT EXISTS orders (
        order_id BIGINT, status STRING, amount DECIMAL(10,2), seq BIGINT, deleted BOOLEAN
    ) USING DELTA TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")

changes = spark.createDataFrame(BATCHES[sys.argv[1]], CHANGES)
newest_first = Window.partitionBy("order_id").orderBy(F.col("seq").desc())
latest = changes.withColumn("n", F.row_number().over(newest_first)).filter("n = 1").drop("n")

(
    DeltaTable.forName(spark, "orders").alias("t")
    .merge(latest.alias("s"), "t.order_id = s.order_id")
    .whenMatchedUpdate(condition="s.seq > t.seq", set={
        "status": "coalesce(s.status, t.status)",
        "amount": "coalesce(s.amount, t.amount)",
        "seq": "s.seq",
        "deleted": "s.op = 'D'",
    })
    .whenNotMatchedInsert(values={
        "order_id": "s.order_id", "status": "s.status", "amount": "s.amount",
        "seq": "s.seq", "deleted": "s.op = 'D'",
    })
    .execute()
)
spark.table("orders").orderBy("order_id").show()
```

Run `python apply_changes.py 1`, then `python apply_changes.py 2`:

```text
+--------+------+------+---+-------+
|order_id|status|amount|seq|deleted|
+--------+------+------+---+-------+
|    1001|  paid| 25.50|  3|  false|
|    1002|placed| 12.00|  2|  false|
|    1003|placed| 60.00|  4|  false|
+--------+------+------+---+-------+

+--------+-------+------+---+-------+
|order_id| status|amount|seq|deleted|
+--------+-------+------+---+-------+
|    1001|shipped| 25.50|  5|  false|
|    1002| placed| 12.00|  6|   true|
|    1003|   paid| 60.00|  7|  false|
+--------+-------+------+---+-------+
```

The first batch inserted 1001 and updated it to paid in the same batch, so the table took only its newest change, sequence 3. MERGE needs that: 2 rows for an order already in the table fail it, as module 4 showed, and 2 rows for a new order are both inserted, a duplicate. The second batch shipped 1001, paid 1003 and deleted 1002, which stays in the table with `deleted` set and the delete's sequence number, 6.

## Changes that come twice or late

Changes can arrive twice, when a delivery is retried, and out of order: Databricks warns that Auto Loader does not promise any order of files. Run `python apply_changes.py 2` again, then `python apply_changes.py late`, which brings 1001's old update to paid and 1002's first insert. Both print the same table as before. Every change carries a sequence number no higher than the row's, so `s.seq > t.seq` turns down the updates, and that makes the job an [idempotent pipeline](https://tecyfy.com/concepts/idempotent-pipeline): run it again on the same input and nothing changes.

The flag is what saves 1002. Remove the row on its delete instead, with `whenMatchedDelete`, and the late insert finds no row to compare with, so `whenNotMatchedInsert` adds order 1002 back as placed. Databricks advises the same 2 rules for streams of files: prefer soft deletes over removing rows, and compare a change's time with the row's before updating. Readers then see only rows where `deleted` is false, through a view, the way [module 5](/data-engineering/course/unity-catalog/personal-data) hid emails.

## The change feed of your own tables

The `orders` table was made with `delta.enableChangeDataFeed = true`, so Delta keeps a log of its own changes, row by row, which the next step of a pipeline can read instead of the whole table. Save this as `changes.py`:

```python
"""What the last batches changed in the orders table, from its change data feed."""
from delta_session import delta_spark

spark = delta_spark("changes")
spark.sql("""
    SELECT _commit_version AS version, _change_type, order_id, status, seq, deleted
    FROM table_changes('orders', 2)
    ORDER BY version, order_id, _change_type DESC
""").show()
```

Run `python changes.py`:

```text
+-------+----------------+--------+-------+---+-------+
|version|    _change_type|order_id| status|seq|deleted|
+-------+----------------+--------+-------+---+-------+
|      2| update_preimage|    1001|   paid|  3|  false|
|      2|update_postimage|    1001|shipped|  5|  false|
|      2| update_preimage|    1002| placed|  2|  false|
|      2|update_postimage|    1002| placed|  6|   true|
|      2| update_preimage|    1003| placed|  4|  false|
|      2|update_postimage|    1003|   paid|  7|  false|
+-------+----------------+--------+-------+---+-------+
```

Version 2 is the second batch: each update appears as the row before it and the row after it. Versions 3 and 4, the rerun and the late batch, changed no rows, so they add nothing to the feed. The feed is not a permanent history. It covers only changes made after it was turned on, and its old versions are removed as the table's old versions are cleaned up, so a pipeline that needs every change copies them into a table of its own.

## On Databricks

The script runs in a notebook without the import and the `delta_spark` line. Databricks also has an automatic change data feed, on Databricks Runtime 19 and above for tables with row tracking, that needs no table property. In a pipeline, the AUTO CDC APIs of Spark Declarative Pipelines apply a change log like this one for you, ordered by a column you name with `sequence_by`, and drop changes that arrive late. Module 7 builds on them.

## Check yourself

1. apply_changes.py removes order 1002 from the table on its delete, instead of setting deleted. A late batch then brings the old insert of 1002, with sequence number 2. What happens?

   A. Nothing, because the sequence number 2 is lower than the delete's 6
   B. The MERGE fails on the missing row
   C. Order 1002 comes back as placed, because no row is left to compare with and the insert branch adds it
   D. The late batch is skipped by the checkpoint

2. A batch holds 2 changes for order 1004, an insert with sequence 8 and an update with sequence 9. What does apply_changes.py do with them?

   A. It keeps only the update, sequence 9, and merges that, because the window keeps the newest change per order
   B. It applies the insert, then the update, in 1 MERGE
   C. The MERGE fails, because 2 rows of the batch match order 1004
   D. It applies only the insert, because inserts run before updates

### Answers

1. C. Once the row is gone, the table no longer knows the delete happened, so the old insert matches nothing and is inserted again. Keeping the row with deleted set to true keeps its sequence number, and the late insert loses the comparison. Databricks recommends soft deletes for Auto Loader streams for this reason.
2. A. row_number over the window ordered by seq, newest first, keeps 1 row per order. Without it, both rows would be inserted, since the order is new, leaving 2 rows for 1004, and the next MERGE that matches 1004 would fail on them. Keeping the newest change also gives the right final state, since the update holds the whole row after it.

## Questions

### What is change data capture?

Change data capture, CDC, is reading a database's changes as a stream of events, each an insert, update or delete of 1 row with a position in the database's log. A copy of the table elsewhere stays current by applying those events in order, instead of copying the whole table each time.

### How do I handle deletes in a CDC pipeline on Delta Lake?

Keep the deleted row and mark it, with a deleted flag and the delete's sequence number, then filter it out where people read the table. A row removed outright cannot stop an older change that arrives late from adding it again, and changes do arrive late or twice, so the mark is what keeps the table right.

## Sources

- [Upsert into a Delta Lake table using merge](https://docs.delta.io/latest/delta-update.html)
- [What is Auto Loader?](https://docs.databricks.com/aws/en/ingestion/cloud-object-storage/auto-loader/)
- [Use change data feed on Databricks](https://docs.databricks.com/aws/en/delta/delta-change-data-feed)
- [Change data feed](https://docs.delta.io/latest/delta-change-data-feed.html)
- [Lakeflow Connect connector concepts](https://docs.databricks.com/aws/en/ingestion/lakeflow-connect/)
- [The AUTO CDC APIs: simplify change data capture with pipelines](https://docs.databricks.com/aws/en/ldp/cdc)