Delta Lake Change Data Feed on Azure Databricks: readChangeFeed, Incremental MERGE, and the Retention Traps

Most Silver-to-Gold jobs I've seen start the same way: re-read the whole Silver table, recompute everything, overwrite Gold. It works until Silver hits a few billion rows and the "daily" job takes four hours to process the 0.1% of rows that actually changed.

Delta Lake's Change Data Feed (CDF) fixes that. It lets you read only the rows that were inserted, updated, or deleted between two table versions, with the type of change attached. This post covers how to turn it on, how to read it in batch and streaming, how to turn a change feed into a correct MERGE, and the retention and schema traps that show up in interviews and in production.

Turning it on

CDF is a table property. You can set it when you create the table or add it later:

-- New table
CREATE TABLE silver.orders (
  order_id    BIGINT,
  customer_id BIGINT,
  status      STRING,
  amount      DECIMAL(18,2),
  updated_at  TIMESTAMP
)
TBLPROPERTIES (delta.enableChangeDataFeed = true);

-- Existing table
ALTER TABLE silver.orders
SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

Two things to know straight away. First, changes are only recorded from the version where you enabled it. You can't go back and get a change feed for history from before that. Second, you can make it the default for new tables in a session with spark.databricks.delta.properties.defaults.enableChangeDataFeed = true, which is handy in a pipeline that creates many tables.

What a change row looks like

When you read the feed, you get the table's columns plus three metadata columns:

• _change_type: one of insert, update_preimage, update_postimage, or delete
• _commit_version: the Delta version that made the change
• _commit_timestamp: when that commit happened

An UPDATE or a matched MERGE produces two rows per changed record: the old values (update_preimage) and the new values (update_postimage). That pair is the most useful part of CDF and also the most common source of bugs, which we'll get to.

Batch reads

changes = (
    spark.read
    .option("readChangeFeed", "true")
    .option("startingVersion", 12)
    .option("endingVersion", 15)      # optional, inclusive
    .table("silver.orders")
)

changes.groupBy("_change_type").count().show()

You can use startingTimestamp and endingTimestamp instead of versions. In SQL, the same thing is a table-valued function:

SELECT _change_type, COUNT(*)
FROM table_changes('silver.orders', 12, 15)
GROUP BY _change_type;

Both bounds are inclusive. If you track "last processed version" yourself, start the next run at last_version + 1, not last_version, or you'll process the boundary commit twice.

To find the current version for your bookkeeping:

latest = spark.sql("DESCRIBE HISTORY silver.orders LIMIT 1").first()["version"]

Streaming reads (usually the better choice)

Hand-rolled version tracking works, but a streaming read lets the checkpoint do the bookkeeping for you:

stream = (
    spark.readStream
    .option("readChangeFeed", "true")
    .option("startingVersion", 12)   # only used on the first run
    .table("silver.orders")
)

startingVersion only matters the first time the query starts with an empty checkpoint. After that, the checkpoint decides where to resume. Combined with trigger(availableNow=True), you get an incremental batch job that processes exactly what changed since the last run and then stops.

Turning changes into a correct MERGE

Here's the trap. Between two runs, the same order_id might have been updated three times, or inserted and then deleted. If you MERGE the raw feed into Gold, you get this error:

Cannot perform Merge as multiple source rows matched and attempted to modify the same target row

Or worse, if you deduplicate carelessly, you apply an old update after a newer one. The fix is to drop the pre-images and keep only the latest change per key:

from pyspark.sql import functions as F
from pyspark.sql.window import Window

def latest_changes(df):
    w = Window.partitionBy("order_id").orderBy(F.col("_commit_version").desc())
    return (
        df.filter(F.col("_change_type") != "update_preimage")
          .withColumn("rn", F.row_number().over(w))
          .filter("rn = 1")
          .drop("rn")
    )

Within a single commit, a key normally has at most one post-image or delete, so _commit_version is enough to order them. Then upsert with foreachBatch:

from delta.tables import DeltaTable

def upsert_to_gold(batch_df, batch_id):
    src = latest_changes(batch_df)
    gold = DeltaTable.forName(spark, "gold.orders_current")
    (
        gold.alias("t")
        .merge(src.alias("s"), "t.order_id = s.order_id")
        .whenMatchedDelete(condition="s._change_type = 'delete'")
        .whenMatchedUpdate(
            condition="s._change_type != 'delete'",
            set={
                "customer_id": "s.customer_id",
                "status": "s.status",
                "amount": "s.amount",
                "updated_at": "s.updated_at",
            },
        )
        .whenNotMatchedInsert(
            condition="s._change_type != 'delete'",
            values={
                "order_id": "s.order_id",
                "customer_id": "s.customer_id",
                "status": "s.status",
                "amount": "s.amount",
                "updated_at": "s.updated_at",
            },
        )
        .execute()
    )

(
    stream.writeStream
    .foreachBatch(upsert_to_gold)
    .option("checkpointLocation", "/Volumes/main/gold/_checkpoints/orders_current")
    .trigger(availableNow=True)
    .start()
)

Notice the condition on the insert: if a key was inserted and deleted inside the same batch, the latest change is a delete, and the row should never reach Gold.

Where pre-images earn their keep: aggregates

For a plain mirror, you throw pre-images away. For aggregates, they're exactly what you need. Say Gold keeps revenue per customer. An update from 100 to 120 shouldn't add 120, it should add 20. Pre-images give you the subtraction:

signed = (
    batch_df
    .withColumn(
        "delta_amount",
        F.when(F.col("_change_type").isin("insert", "update_postimage"), F.col("amount"))
         .when(F.col("_change_type").isin("delete", "update_preimage"), -F.col("amount"))
    )
    .groupBy("customer_id")
    .agg(F.sum("delta_amount").alias("delta_amount"))
)

Then MERGE that into the aggregate table with t.revenue = t.revenue + s.delta_amount. You update a handful of customers instead of re-aggregating the whole table. Be careful about idempotency here: an additive merge applied twice double-counts. foreachBatch can replay a batch after a failure, so either make the write idempotent with Delta's txnAppId/txnVersion options (on appends) or keep the aggregate recomputable from Silver as a fallback.

The traps interviewers ask about

Retention. Change data lives in a _change_data folder inside the table (for some operations, like insert-only appends, Delta derives the changes from the regular data files instead). Those files are removed by VACUUM like any other, under delta.deletedFileRetentionDuration, which defaults to 7 days. If your consumer is down for two weeks, the versions it needs may be gone, and the read fails. Size the retention to your worst realistic outage, not your normal schedule.

Reading before CDF was enabled. Asking for a version from before you turned it on is an error, not an empty result. Start from the version where the property was set (you'll find it in DESCRIBE HISTORY as a SET TBLPROPERTIES operation).

Schema changes. Additive changes such as a new column are fine. Non-additive changes on tables with column mapping, like renaming or dropping a column, can block batch CDF reads over a range that spans the change. Plan those changes, and restart the downstream stream deliberately afterwards.

Overwrites. A full overwrite of the source shows up as deletes of every old row and inserts of every new row. Your incremental job then processes the entire table. If an upstream job overwrites "to be safe", CDF won't save you; switching it to MERGE or replaceWhere will.

Cost. Enabling CDF adds a little write overhead for updates, deletes, and merges, because Delta writes the change files too. For tables nobody reads incrementally, leave it off.

Interview questions to practise

1. What are the four _change_type values, and which operations produce update_preimage?
2. Why can't you MERGE a raw change feed straight into a target table?
3. A CDF consumer was paused for 10 days and now fails on start. What probably happened?
4. When would you keep pre-images instead of dropping them?
5. How is reading CDF as a stream different from tracking startingVersion yourself?

If you can explain question 2 with a concrete example of three updates to one key, you've got the core idea.

---

If you want more practice like this, there's a free 10-question sample of Azure Databricks and PySpark interview questions with worked answers: free sample. The full set is in the Azure Databricks & PySpark Interview Pack.