Delta Lake & Change Data Feed: The Modern Data Engineer’s Secret Weapon

How Delta Lake solves real-world data reliability problems — and how Change Data Feed takes it to the next level

The Problem With Raw Data Lakes

Imagine you’re a data engineer at a mid-sized e-commerce company. Every day, millions of rows of order, user, and inventory data land in your data lake — a massive folder of Parquet files on S3 or ADLS. Life seems good.

Then your manager walks in:

“Hey, a customer said their order status is wrong in the dashboard. Can you fix just that one record?”

You stare at your screen. You have no idea how to update a single row in a Parquet file without rewriting the entire partition. And even if you did — what happens to the downstream pipelines reading that same data right now?

This is the everyday reality of working with raw data lakes. And this is exactly why Delta Lake was born.

What Is Delta Lake?

Delta Lake is an open-source storage layer that brings ACID transactions, schema enforcement, and time travel to your existing data lake (S3, ADLS, GCS, HDFS).

Think of it as giving your data lake a proper backbone — the reliability guarantees you’d expect from a database, but at data lake scale.

Delta Lake sits on top of your existing cloud storage. You keep the cost benefits of a data lake; you gain the reliability of a data warehouse.

The Core Features

Feature What it means for you ACID Transactions No more partial writes or corrupt reads Schema Enforcement Bad data gets rejected at the door Time Travel Query your data as it looked 7 days ago Upserts & Deletes MERGE, UPDATE, DELETE — like a real database Scalable Metadata Handles billions of files efficiently Unified Batch + Streaming One table and one format for both workloads

Why Delta Lake Matters Day-to-Day

Let walk through three scenarios every data team faces regularly.

Scenario 1: GDPR Delete Request

A user requests data deletion. In a raw Parquet lake, you’d need to rewrite every partition that contains their data — a painful, error-prone process.

With Delta Lake:

DELETE FROM orders WHERE user_id = 'USR_4892';

Done. The transaction is logged. The file is updated. Downstream readers see a consistent view.

Scenario 2: Late-Arriving Data

A payment processor sends corrected transaction data two days late. You need to update records without breaking anything.

With Delta Lake:

MERGE INTO transactions AS target
USING late_corrections AS source
ON target.txn_id = source.txn_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

Scenario 3: “What did our data look like last Tuesday?”

Your ML model trained on last Tuesday’s snapshot is behaving oddly. You need to reproduce that exact dataset.

With Delta Lake (Time Travel):

df = spark.read.format("delta") \
    .option("timestampAsOf", "2024-11-05") \
    .load("/data/transactions")

No snapshots. No extra storage. Just the transaction log doing its magic.

Where Does Change Data Feed (CDF) Come In?

Delta Lake already solves a lot. But there’s still a missing piece.

Suppose you have a customers table with 500 million rows. Downstream, you have:

• A feature store that needs updated customer attributes for ML models
• A CDC pipeline syncing changes to Elasticsearch for search
• A data mart in Redshift that needs to stay in sync

Every hour, maybe 50,000 rows change. But your downstream jobs have no clean way to ask: “Give me only what changed since the last time I ran.”

Without CDF, they have two ugly options:

1. Full table scan — Read all 500M rows, compare, find the 50K that changed. Expensive and slow.
2. Timestamp-based filtering — Fragile, misses deletes, doesn’t capture intermediate updates.

Change Data Feed solves this elegantly.

What Is Delta Lake Change Data Feed (CDF)?

Change Data Feed (also called Change Data Capture in Delta Lake) is a feature that automatically tracks row-level changes — inserts, updates, and deletes — on a Delta table.

When CDF is enabled, Delta Lake maintains a hidden _change_data folder alongside your table data. Every DML operation appends a structured record of what changed, what the old value was, what the new value is, and what type of change it was.

The Four Change Types

_change_type Description insert A new row was added update_preimage The row's value before the update update_postimage The row's value after the update delete A row was removed

Implementing Delta Lake CDF: A Complete Example

Let’s build a realistic pipeline. We have a customers table, and we want a downstream customer_features table to stay in sync efficiently.

Step 1: Environment Setup

# Install Delta Lake (for local/Databricks-free testing)
# pip install delta-spark pyspark

from pyspark.sql import SparkSession
from delta import configure_spark_with_delta_pip

builder = SparkSession.builder \
    .appName("DeltaCDF-Demo") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", 
            "org.apache.spark.sql.delta.catalog.DeltaCatalog")

spark = configure_spark_with_delta_pip(builder).getOrCreate()

Step 2: Create the Source Table with CDF Enabled

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, TimestampType
from delta.tables import DeltaTable
from datetime import datetime

# Sample customer data
data = [
    (1, "Alice",   "alice@example.com",   "Mumbai",    "Gold",   datetime(2024, 1, 1)),
    (2, "Bob",     "bob@example.com",     "Delhi",     "Silver", datetime(2024, 1, 2)),
    (3, "Charlie", "charlie@example.com", "Bangalore", "Bronze", datetime(2024, 1, 3)),
    (4, "Diana",   "diana@example.com",   "Pune",      "Gold",   datetime(2024, 1, 4)),
    (5, "Eve",     "eve@example.com",     "Chennai",   "Silver", datetime(2024, 1, 5)),
]

schema = StructType([
    StructField("customer_id",  IntegerType(),  False),
    StructField("name",         StringType(),   True),
    StructField("email",        StringType(),   True),
    StructField("city",         StringType(),   True),
    StructField("tier",         StringType(),   True),
    StructField("signup_date",  TimestampType(),True),
])

df = spark.createDataFrame(data, schema)

# Write with CDF enabled — this is the key property!
df.write.format("delta") \
    .option("delta.enableChangeDataFeed", "true") \
    .mode("overwrite") \
    .save("/delta/customers")

print("Customers table created with CDF enabled.")

Step 3: Make Some Changes

Now let’s simulate real-world changes — an update, an upsert, and a delete.

customers_table = DeltaTable.forPath(spark, "/delta/customers")

# --- UPDATE: Alice got promoted to Platinum ---
customers_table.update(
    condition = "customer_id = 1",
    set        = {"tier": "'Platinum'"}
)
print("Updated Alice's tier to Platinum")

# --- MERGE (Upsert): New customer + update Bob's city ---
upsert_data = [
    (2, "Bob",   "bob@example.com",   "Hyderabad", "Silver", datetime(2024, 1, 2)),  # city changed
    (6, "Frank", "frank@example.com", "Kolkata",   "Bronze", datetime(2024, 11, 1)), # new customer
]
upsert_df = spark.createDataFrame(upsert_data, schema)

customers_table.alias("target").merge(
    upsert_df.alias("source"),
    "target.customer_id = source.customer_id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()
print("Merged updates — Bob relocated, Frank joined")

# --- DELETE: Eve churned ---
customers_table.delete("customer_id = 5")
print("Deleted Eve's record (churn)")

Step 4: Read the Change Data Feed

This is the magic moment. Let’s read exactly what changed.

# Read CDF changes starting from version 1 (after initial load)
changes_df = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", 1) \
    .load("/delta/customers")

changes_df.select(
    "customer_id", "name", "tier", "city",
    "_change_type", "_commit_version", "_commit_timestamp"
).orderBy("_commit_version", "customer_id").show(truncate=False)

Output:

+-----------+-------+----------+-----------+------------------+----------------+-------------------+
|customer_id|name   |tier      |city       |_change_type      |_commit_version |_commit_timestamp  |
+-----------+-------+----------+-----------+------------------+----------------+-------------------+
|1          |Alice  |Gold      |Mumbai     |update_preimage   |1               |2024-11-06 10:01:00|
|1          |Alice  |Platinum  |Mumbai     |update_postimage  |1               |2024-11-06 10:01:00|
|2          |Bob    |Silver    |Delhi      |update_preimage   |2               |2024-11-06 10:02:00|
|2          |Bob    |Silver    |Hyderabad  |update_postimage  |2               |2024-11-06 10:02:00|
|5          |Eve    |Silver    |Chennai    |delete            |3               |2024-11-06 10:03:00|
|6          |Frank  |Bronze    |Kolkata    |insert            |2               |2024-11-06 10:02:00|
+-----------+-------+----------+-----------+------------------+----------------+-------------------+

Every insert, update (with before/after), and delete — captured automatically. No custom triggers. No Debezium. No Kafka connectors needed.

Step 5: Build an Incremental Downstream Pipeline

Now let’s use CDF to build an efficient incremental sync to a downstream customer_features table.

# Simulated state: last time we synced, the table was at version 0
last_processed_version = 0

def get_latest_version(table_path):
    dt = DeltaTable.forPath(spark, table_path)
    return dt.history(1).select("version").collect()[0][0]

def run_incremental_sync(source_path, target_path, from_version):
    """
    Reads only changed rows from source since from_version,
    and applies them to the target table.
    """
    current_version = get_latest_version(source_path)

    if from_version >= current_version:
        print("No new changes to process.")
        return current_version

    print(f"Processing versions {from_version + 1} → {current_version}")

    # Read the change feed
    cdf = spark.read.format("delta") \
        .option("readChangeFeed", "true") \
        .option("startingVersion", from_version + 1) \
        .load(source_path)

    # We only care about the latest state of each customer
    # Keep only inserts and update_postimage (final state)
    latest_changes = cdf.filter(
        "(_change_type = 'insert') OR (_change_type = 'update_postimage')"
    ).drop("_change_type", "_commit_version", "_commit_timestamp")

    deleted_ids = cdf.filter("_change_type = 'delete'") \
                     .select("customer_id") \
                     .distinct()

    # Apply to target — upsert changed rows
    if DeltaTable.isDeltaTable(spark, target_path):
        target = DeltaTable.forPath(spark, target_path)

        # Handle deletes
        if deleted_ids.count() > 0:
            deleted_list = [row.customer_id for row in deleted_ids.collect()]
            target.delete(f"customer_id IN ({','.join(map(str, deleted_list))})")
            print(f"Deleted {len(deleted_list)} records")

        # Handle upserts
        if latest_changes.count() > 0:
            target.alias("t").merge(
                latest_changes.alias("s"),
                "t.customer_id = s.customer_id"
            ).whenMatchedUpdateAll() \
             .whenNotMatchedInsertAll() \
             .execute()
            print(f"Upserted {latest_changes.count()} records")
    else:
        # First run — just write
        latest_changes.write.format("delta").save(target_path)
        print(f"Initial write: {latest_changes.count()} records")

    return current_version

# Run the sync
new_checkpoint = run_incremental_sync(
    source_path    = "/delta/customers",
    target_path    = "/delta/customer_features",
    from_version   = last_processed_version
)

print(f"Sync complete. New checkpoint version: {new_checkpoint}")

Step 6: CDF With Streaming (Bonus!)

CDF works beautifully with Spark Structured Streaming for near-real-time pipelines.

# Streaming read of changes — processes new commits as they arrive
streaming_changes = spark.readStream.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", "latest") \
    .load("/delta/customers")

# Only process inserts and final update states
processed_stream = streaming_changes.filter(
    "(_change_type = 'insert') OR (_change_type = 'update_postimage')"
)

# Write to a streaming sink (another Delta table, Kafka, etc.)
query = processed_stream.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/delta/checkpoints/customer_sync") \
    .start("/delta/customer_features_stream")

query.awaitTermination(30)  # run for 30 seconds in demo
print("Streaming CDF pipeline running")

Key Configuration Reference

Enabling CDF

-- At table creation
CREATE TABLE customers (...)
USING DELTA
TBLPROPERTIES (delta.enableChangeDataFeed = true);

-- On an existing table
ALTER TABLE customers
SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

Reading CDF: All Options

# By version range
spark.read.format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", 5)
    .option("endingVersion", 10)   # optional
    .load("/delta/table")

# By timestamp range
spark.read.format("delta")
    .option("readChangeFeed", "true")
    .option("startingTimestamp", "2024-11-01 00:00:00")
    .option("endingTimestamp",   "2024-11-06 00:00:00")  # optional
    .load("/delta/table")

When Should You Use CDF?

Use CDF when:

• You have large tables (100M+ rows) with small daily change rates (<5%)
• You need to sync changes to ML feature stores, search indexes, or downstream warehouses
• You’re building event-driven pipelines that react to data changes
• You need audit trails of who changed what and when
• You want to replace expensive CDC tools (Debezium + Kafka) for Delta-to-Delta pipelines

CDF is not ideal when:

• Your entire table changes every run (full refreshes) — overhead with no benefit
• Your table has very high write frequency (millions of small transactions/second) — consider Kafka-first architectures
• Storage cost is extremely constrained — CDF files are retained until VACUUM runs

CDF vs Traditional CDC: A Comparison

Traditional CDC (Debezium + Kafka) Delta Lake CDF Setup complexity High — connectors, schemas, Kafka cluster Low — one table property Latency Near real-time (ms) Near real-time (streaming) or batch Pre/post image Depends on DB config Built-in always Works with Source databases (Postgres, MySQL) Delta Lake tables only Cost Kafka infrastructure overhead Storage for change files only Best for DB → Data Lake ingestion Delta → Delta / downstream sync

Tags: #DataEngineering #DeltaLake #ApacheSpark #BigData #DataLakehouse #CDC #PySpark #Databricks