Delta change data feed on Databricks: incremental silver-to-gold propagation without full rescans

Rebuilding a gold table from a 5 TB silver table every hour, because a few thousand rows changed, is expensive. Change data feed (CDF) records row-level changes on a Delta table so downstream jobs process only what changed, including updates and deletes.

#Enable it

ALTER TABLE sales_prod.silver.customers SET TBLPROPERTIES (delta.enableChangeDataFeed = true);
-- or at creation
CREATE TABLE sales_prod.silver.orders (...) TBLPROPERTIES (delta.enableChangeDataFeed = true);

CDF only records changes after it is enabled. Note the version at which you enabled it:

DESCRIBE HISTORY sales_prod.silver.customers LIMIT 5;

#What a change looks like

Reading CDF adds three columns:

SELECT customer_id, email, _change_type, _commit_version
FROM table_changes('sales_prod.silver.customers', 120, 125)
ORDER BY _commit_version;

#Batch reads by version

changes = (spark.read
    .option("readChangeFeed", "true")
    .option("startingVersion", last_processed + 1)
    .option("endingVersion", current_version)
    .table("sales_prod.silver.customers"))

Store last_processed in a small control table, and update it in the same job run after the downstream write succeeds.

#Streaming reads (simplest)

(spark.readStream
    .option("readChangeFeed", "true")
    .option("startingVersion", 125)       # only on first start; afterwards the checkpoint decides
    .table("sales_prod.silver.customers")
 .writeStream
    .foreachBatch(apply_changes)
    .option("checkpointLocation", "/Volumes/sales_prod/_checkpoints/gold_customers")
    .trigger(availableNow=True)
    .start())

availableNow processes everything new and stops, which fits scheduled jobs.

#Apply changes with MERGE

A version can contain several changes per key; keep only the latest per key, and drop pre-images:

from pyspark.sql import Window, functions as F

def apply_changes(batch_df, batch_id):
    w = Window.partitionBy("customer_id").orderBy(F.col("_commit_version").desc())
    latest = (batch_df.filter("_change_type != 'update_preimage'")
              .withColumn("rn", F.row_number().over(w)).filter("rn = 1").drop("rn"))
    latest.createOrReplaceTempView("changes")
    batch_df.sparkSession.sql("""
      MERGE INTO sales_prod.gold.dim_customer t
      USING changes s ON t.customer_id = s.customer_id
      WHEN MATCHED AND s._change_type = 'delete' THEN DELETE
      WHEN MATCHED THEN UPDATE SET t.email = s.email, t.country = s.country, t.updated_at = s._commit_timestamp
      WHEN NOT MATCHED AND s._change_type != 'delete' THEN
        INSERT (customer_id, email, country, updated_at) VALUES (s.customer_id, s.email, s.country, s._commit_timestamp)
    """)

Warning

Without the dedupe step, MERGE fails with "multiple source rows matched" whenever a key changed twice in the batch. Without the preimage filter, updates are applied twice in the wrong order.

For aggregated gold tables (sums by day), recompute only the affected keys: collect the distinct dates or customers from the changes, then rebuild those slices with replaceWhere or MERGE.

#Retention and gaps

CDF data lives in the table's _change_data folder and is removed by VACUUM along with old files, following the table's retention. If a downstream job is down longer than retention:

DeltaFileNotFoundException / "change data for version X is not available"

Recovery: do a one-time full rebuild of the downstream table, then restart CDF from the current version. Keep delta.deletedFileRetentionDuration (and delta.logRetentionDuration) longer than your worst-case outage.

#Schema changes

• Adding columns: CDF continues; new columns appear in later versions.
• Column renames/drops with column mapping: streaming reads of CDF may fail across non-additive schema changes. Restart with a new checkpoint from the version after the change, or read in batch mode around it. Check the current documentation for your runtime's support.

#Operations that look like big changes

Avoid overwriting CDF-enabled tables in full; use MERGE or replaceWhere on a slice.

#When not to use CDF

• Append-only tables: a plain streaming read of the Delta table is simpler.
• Tables you fully rebuild each run: changes are the whole table anyway.
• Very small tables: full rebuild is cheaper than the bookkeeping.

#Prevent it next time

• Control table or checkpoint per consumer, monitored for lag in versions.
• Retention settings documented on every CDF table, longer than the longest planned outage.
• Upstream writers use MERGE, not overwrite, when consumers rely on CDF.