Delta Change Data Feed Complete Guide: CDC, Incremental Processing, Downstream Integration

Change Data Feed (CDF) automatically records changes (INSERT / UPDATE / DELETE) to a Delta Table and efficiently propagates "what changed" to downstream pipelines and systems. It is commonly used for incremental processing between Silver and Gold layers, and for integration with external systems.

This article walks through enabling CDF, the meaning of each _change_type value, how to read CDF data in batch and streaming modes, the foreachBatch + MERGE pattern for downstream propagation, how CDF relates to DLT APPLY CHANGES INTO, and the points the exam tends to ask about.

Enabling CDF

CDF is enabled per table. You can set it when creating a new table or by changing properties on an existing one. Only changes that occur after CDF is enabled are recorded — history from before enabling is not available.

-- 新規テーブル作成時に有効化
CREATE TABLE silver.orders (
  order_id BIGINT,
  customer_id BIGINT,
  amount DECIMAL(10,2),
  status STRING,
  updated_at TIMESTAMP
)
TBLPROPERTIES (delta.enableChangeDataFeed = true);

-- 既存テーブルに対して有効化
ALTER TABLE silver.orders
SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

-- ワークスペース全体でデフォルト有効化
SET spark.databricks.delta.properties.defaults.enableChangeDataFeed = true;

The _change_type Values

Data read via CDF automatically includes three metadata columns. The most important one is _change_type, which indicates the kind of change.

For UPDATEs, both a preimage (before) and a postimage (after) row are recorded. This ability to trace exactly "what changed into what" is CDF's strength. Two additional metadata columns are also added: _commit_version (the Delta version number) and _commit_timestamp (the commit time).

Reading from CDF

Batch read: the table_changes() function

-- バージョン範囲で変更を取得
SELECT * FROM table_changes('silver.orders', 5, 10);

-- タイムスタンプ範囲で変更を取得
SELECT * FROM table_changes('silver.orders', '2026-03-01', '2026-03-27');

-- 特定の_change_typeだけフィルタ
SELECT * FROM table_changes('silver.orders', 5)
WHERE _change_type IN ('insert', 'update_postimage');

Streaming read: the readChangeFeed option

# バージョン指定で開始
cdf_stream = (spark.readStream
  .format("delta")
  .option("readChangeFeed", "true")
  .option("startingVersion", 5)
  .table("silver.orders")
)

# タイムスタンプ指定で開始
cdf_stream = (spark.readStream
  .format("delta")
  .option("readChangeFeed", "true")
  .option("startingTimestamp", "2026-03-01")
  .table("silver.orders")
)

Incremental Propagation to Downstream Pipelines (foreachBatch + MERGE)

The most common use of CDF is incremental processing between Silver and Gold. Instead of fully rebuilding the Gold aggregate table every run, you fetch only the changes via CDF and apply them with MERGE.

def update_gold(batch_df, batch_id):
    updates = batch_df.filter(
        "_change_type IN ('insert', 'update_postimage')"
    )
    deletes = batch_df.filter("_change_type = 'delete'")

    updates.createOrReplaceTempView("updates")
    deletes.createOrReplaceTempView("deletes")

    # UPSERT: 新規挿入と更新後の状態を反映
    spark.sql("""
      MERGE INTO gold.order_summary t
      USING updates s ON t.order_id = s.order_id
      WHEN MATCHED THEN UPDATE SET *
      WHEN NOT MATCHED THEN INSERT *
    """)

    # DELETE: 削除された行をGoldからも削除
    spark.sql("""
      MERGE INTO gold.order_summary t
      USING deletes s ON t.order_id = s.order_id
      WHEN MATCHED THEN DELETE
    """)

(spark.readStream
  .format("delta")
  .option("readChangeFeed", "true")
  .table("silver.orders")
  .writeStream
  .foreachBatch(update_gold)
  .option("checkpointLocation", "/checkpoint/gold_orders")
  .trigger(availableNow=True)
  .start()
)

trigger(availableNow=True) runs the stream in a batch-like fashion, processing all available CDF data and then stopping the stream automatically. It pairs well with scheduled jobs.

Integration with External Systems

CDF is also useful for incremental data integration with external systems. Sending only the changes instead of resending the full dataset dramatically reduces network and processing costs.

• Incremental loads into Snowflake / Redshift
• Index updates for Elasticsearch / OpenSearch
• Cache updates for Redis / DynamoDB
• Publishing change events to Kafka

Relationship with DLT APPLY CHANGES INTO

Delta Live Tables (DLT) APPLY CHANGES INTO is conceptually similar to CDF but lives at a different layer. CDF is a change-recording feature on the Delta Table side, while APPLY CHANGES INTO is a CDC-ingestion syntax on the DLT pipeline side.

-- DLT APPLY CHANGES INTO の例
APPLY CHANGES INTO LIVE.silver_orders
FROM STREAM(LIVE.bronze_orders_cdc)
KEYS (order_id)
SEQUENCE BY updated_at
COLUMNS * EXCEPT (_rescued_data)
STORED AS SCD TYPE 1;

Retention and VACUUM

CDF data is stored in the Delta Table's _change_data directory. Running VACUUM also removes CDF data that falls outside the retention window.

-- VACUUM実行（7日間保持）
VACUUM silver.orders RETAIN 168 HOURS;

-- 下流パイプラインの処理遅延を考慮して保持期間を長めに設定
ALTER TABLE silver.orders
SET TBLPROPERTIES (
  'delta.deletedFileRetentionDuration' = 'interval 30 days'
);

Points the Exam Tests

• How to enable CDF: TBLPROPERTIES (delta.enableChangeDataFeed = true)
• The four _change_type values: insert / update_preimage / update_postimage / delete
• Streaming reads via the readChangeFeed option
• Batch reads via the table_changes() function
• Changes from before CDF was enabled cannot be retrieved
• UPDATEs record two rows: preimage and postimage
• APPLY CHANGES INTO is DLT pipeline syntax — a separate layer from CDF

Check Your Understanding

Frequently Asked Questions