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

Part 1 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, this time with deletes.

What a database sends

A database keeps a log of the changes it makes. Change data captureChange 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. 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:

"""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:

+--------+------+------+---+-------+
|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 pipelineIdempotent Pipelines and Exactly-OnceA pipeline step is idempotent when running it twice gives the same result as running it once. That is what makes retries safe, and it is how systems that deliver a message more than once still count it exactly once.: 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 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:

"""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:

+-------+----------------+--------+-------+---+-------+
|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.