CDC — Change Data Capture: A Complete Production Guide with PySpark

Change Data Capture (CDC) is a design pattern used to identify and track row-level changes (inserts, updates, deletes) in a source database and propagate those changes downstream to a target system — such as a data lake, data warehouse, or analytics platform — without performing expensive full-table reloads.

CDC is the backbone of modern real-time data pipelines and is one of the most frequently asked topics in Data Engineering interviews.

What is CDC?

In traditional ETL pipelines, you load an entire source table every day into your warehouse (a "full dump"). If the table has 500 million rows but only 10,000 rows changed today, you're needlessly re-reading and re-writing 499,990,000 unchanged rows.

CDC eliminates this waste by capturing only the rows that actually changed since the last extraction.

How CDC Captures Changes

There are three primary mechanisms by which CDC systems detect changes:

The CDC Event Structure

Regardless of the mechanism, a CDC event typically carries this information:

{
  "operation": "U",             ← I = Insert, U = Update, D = Delete
  "timestamp": "2026-05-31T12:00:05Z",
  "before": {                   ← The row BEFORE the change (for U and D)
    "customer_id": "C101",
    "name": "Amit Kumar",
    "city": "Mumbai"
  },
  "after": {                    ← The row AFTER the change (for I and U)
    "customer_id": "C101",
    "name": "Amit Kumar",
    "city": "Bangalore"
  }
}

Why is CDC Used?

1. Efficiency: Instead of copying 500M rows daily, you process only the 10K rows that changed. This reduces compute costs by 99%+ for large tables.
2. Near Real-Time Freshness: Log-based CDC can stream changes within seconds, enabling real-time dashboards, ML feature stores, and operational analytics.
3. Reduced Source Load: Full table dumps put heavy read pressure on the production database. CDC reads from the transaction log (a sequential file), causing minimal load on the live database.
4. Auditability: CDC events include before and after images, creating a natural audit trail of every change ever made.
5. Data Lake Sync: CDC is the standard pattern for keeping a data lake (S3/HDFS) in sync with transactional databases (MySQL, PostgreSQL, Oracle) without nightly downtime windows.

CDC vs SCD — How Are They Different?

This is one of the most common interview confusion points. CDC and SCD are NOT alternatives — they solve different problems at different layers of the pipeline.

How They Work Together

Source DB ──[CDC]──► Raw/Staging Zone ──[SCD-2 Logic]──► Dimension Table

  MySQL            Debezium extracts        PySpark applies           dim_customer
  (OLTP)           change events            SCD-2 merge logic         (Data Warehouse)

CDC captures the changes. SCD decides how to store them. You can use CDC without SCD (e.g., just overwrite the target table), and you can use SCD without CDC (e.g., compare full snapshots). But in production, they are almost always used together.

CDC Pipeline Architecture

The following diagram illustrates the end-to-end CDC pipeline: source database transaction logs are captured by a CDC tool (Debezium/AWS DMS), streamed through Kafka topics, consumed by PySpark, and applied as UPSERT/DELETE operations to the target data lake.

Common CDC Tools in Production

Step-by-Step Implementation: Timestamp-Based CDC in Pure PySpark

This implementation uses no Delta Lake, no Hudi, no Iceberg — just plain PySpark and Parquet files. This is the approach you'd use in environments without lakehouse table formats.

Step 1: Create the Initial Target Table (First Full Load)

On day one, you perform a full extract from the source database and write it to your data lake as the baseline.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import *

spark = SparkSession.builder \
    .appName("CDC-PySpark-Implementation") \
    .getOrCreate()

# -------------------------------------------------------
# Simulate the source database table (Day 1 — Initial Full Load)
# -------------------------------------------------------
initial_data = [
    ("C101", "Amit Kumar",  "Mumbai",    "Gold",   "2026-05-01 10:00:00"),
    ("C102", "Priya Rao",   "Pune",      "Silver", "2026-05-01 10:00:00"),
    ("C103", "Mukesh Sen",  "Delhi",     "Gold",   "2026-05-01 10:00:00"),
    ("C104", "Neha Gupta",  "Hyderabad", "Bronze", "2026-05-01 10:00:00"),
]

source_schema = StructType([
    StructField("customer_id", StringType(), False),
    StructField("name", StringType(), True),
    StructField("city", StringType(), True),
    StructField("tier", StringType(), True),
    StructField("updated_at", StringType(), False),
])

initial_df = spark.createDataFrame(initial_data, source_schema) \
    .withColumn("updated_at", F.to_timestamp("updated_at"))

# Write the full initial load to the target data lake path
TARGET_PATH = "/data/lake/customers"
initial_df.write.mode("overwrite").parquet(TARGET_PATH)

print("✅ Initial full load completed:")
initial_df.show(truncate=False)

Initial Target Table:

+-----------+------------+---------+------+-------------------+
|customer_id|name        |city     |tier  |updated_at         |
+-----------+------------+---------+------+-------------------+
|C101       |Amit Kumar  |Mumbai   |Gold  |2026-05-01 10:00:00|
|C102       |Priya Rao   |Pune     |Silver|2026-05-01 10:00:00|
|C103       |Mukesh Sen  |Delhi    |Gold  |2026-05-01 10:00:00|
|C104       |Neha Gupta  |Hyderabad|Bronze|2026-05-01 10:00:00|
+-----------+------------+---------+------+-------------------+

Step 2: Simulate the CDC Change Events (Day 2 Batch)

On day 2, the source database has received several changes. In timestamp-based CDC, we query the source for all rows where updated_at > last_extraction_time. We also simulate a delete by using a _cdc_operation marker column.

# -------------------------------------------------------
# Simulate Day 2 CDC events arriving from the source
# -------------------------------------------------------
# In production, this could come from:
#   - A Kafka topic consumed by Spark Structured Streaming
#   - An AWS DMS output landing zone in S3
#   - A direct JDBC query with WHERE updated at > last run

cdc_events = [
    ("C101", "Amit Kumar",  "Bangalore", "Gold",     "2026-05-02 09:30:00", "U"),  # UPDATE: city changed
    ("C103", "Mukesh Sen",  "Delhi",     "Platinum", "2026-05-02 11:15:00", "U"),  # UPDATE: tier changed
    ("C104", "Neha Gupta",  "Hyderabad", "Bronze",   "2026-05-02 14:00:00", "D"),  # DELETE: customer removed
    ("C105", "Ravi Sharma", "Chennai",   "Silver",   "2026-05-02 08:45:00", "I"),  # INSERT: brand new customer
]

cdc_schema = StructType([
    StructField("customer_id", StringType(), False),
    StructField("name", StringType(), True),
    StructField("city", StringType(), True),
    StructField("tier", StringType(), True),
    StructField("updated_at", StringType(), False),
    StructField("_cdc_operation", StringType(), False),  # I=Insert, U=Update, D=Delete
])

cdc_df = spark.createDataFrame(cdc_events, cdc_schema) \
    .withColumn("updated_at", F.to_timestamp("updated_at"))

print("📥 Day 2 CDC Events Received:")
cdc_df.show(truncate=False)

CDC Events:

+-----------+------------+---------+--------+-------------------+--------------+
|customer_id|name        |city     |tier    |updated_at         |_cdc_operation|
+-----------+------------+---------+--------+-------------------+--------------+
|C101       |Amit Kumar  |Bangalore|Gold    |2026-05-02 09:30:00|U             |
|C103       |Mukesh Sen  |Delhi    |Platinum|2026-05-02 11:15:00|U             |
|C104       |Neha Gupta  |Hyderabad|Bronze  |2026-05-02 14:00:00|D             |
|C105       |Ravi Sharma |Chennai  |Silver  |2026-05-02 08:45:00|I             |
+-----------+------------+---------+--------+-------------------+--------------+

Step 3: Handle Duplicate / Out-of-Order CDC Events

In real-world CDC pipelines, you can receive duplicate events (e.g., Kafka at-least-once delivery) or multiple changes for the same key in a single batch (e.g., a customer updated their city twice in one day). Before applying changes, you must deduplicate and keep only the latest event per key.

from pyspark.sql.window import Window

# -------------------------------------------------------
# Deduplicate: Keep only the LATEST event per customer id
# -------------------------------------------------------
dedup_window = Window.partitionBy("customer_id").orderBy(F.col("updated_at").desc())

cdc_deduped = cdc_df.withColumn("row_num", F.row_number().over(dedup_window)) \
    .filter(F.col("row_num") == 1) \
    .drop("row_num")

print("🧹 Deduplicated CDC Events (latest per customer):")
cdc_deduped.show(truncate=False)

Step 4: Separate CDC Events by Operation Type

Split the deduplicated CDC stream into three logical groups based on the operation type:

# -------------------------------------------------------
# Split CDC events into INSERT, UPDATE, DELETE streams
# -------------------------------------------------------
inserts_df = cdc_deduped.filter(F.col("_cdc_operation") == "I").drop("_cdc_operation")
updates_df = cdc_deduped.filter(F.col("_cdc_operation") == "U").drop("_cdc_operation")
deletes_df = cdc_deduped.filter(F.col("_cdc_operation") == "D").drop("_cdc_operation")

print(f"📊 Event Breakdown: {inserts_df.count()} INSERTs, {updates_df.count()} UPDATEs, {deletes_df.count()} DELETEs")

Output:

📊 Event Breakdown: 1 INSERTs, 2 UPDATEs, 1 DELETEs

Step 5: Load the Existing Target Table

Read the current state of the target data lake table that was created during the initial full load:

# -------------------------------------------------------
# Read the existing target table from the data lake
# -------------------------------------------------------
existing_df = spark.read.parquet(TARGET_PATH)

print("📦 Current Target Table (before applying CDC):")
existing_df.show(truncate=False)

Step 6: Apply CDC Changes — The Core UPSERT + DELETE Logic

This is the heart of the CDC pipeline. We reconstruct the target table by:

1. Removing rows that were either UPDATED or DELETED from the existing table.
2. Adding the new versions from the UPDATE and INSERT streams.

# -------------------------------------------------------
# Collect all customer ids that are affected by this CDC batch
# (either updated or deleted)
# -------------------------------------------------------
affected_keys = updates_df.select("customer_id") \
    .union(deletes_df.select("customer_id"))

# -------------------------------------------------------
# Step A: REMOVE affected rows from the existing target
# Left anti-join keeps only rows whose customer id is NOT in the affected set
# -------------------------------------------------------
surviving_df = existing_df.join(
    affected_keys,
    on="customer_id",
    how="left_anti"
)

print("🔍 Surviving rows (unchanged + not deleted):")
surviving_df.show(truncate=False)

# -------------------------------------------------------
# Step B: UNION the surviving rows + updated rows + new inserts
# -------------------------------------------------------
final_df = surviving_df \
    .unionByName(updates_df) \
    .unionByName(inserts_df)

print("✅ Final Target Table AFTER applying CDC batch:")
final_df.orderBy("customer_id").show(truncate=False)

Expected Final Output:

+-----------+------------+---------+--------+-------------------+
|customer_id|name        |city     |tier    |updated_at         |
+-----------+------------+---------+--------+-------------------+
|C101       |Amit Kumar  |Bangalore|Gold    |2026-05-02 09:30:00|  ← UPDATED (was Mumbai)
|C102       |Priya Rao   |Pune     |Silver  |2026-05-01 10:00:00|  ← UNCHANGED
|C103       |Mukesh Sen  |Delhi    |Platinum|2026-05-02 11:15:00|  ← UPDATED (was Gold tier)
|C105       |Ravi Sharma |Chennai  |Silver  |2026-05-02 08:45:00|  ← INSERTED (new customer)
+-----------+------------+---------+--------+-------------------+

Notice:

• C101: City changed from Mumbai → Bangalore ✅
• C102: Untouched — no CDC event for this customer ✅
• C103: Tier changed from Gold → Platinum ✅
• C104: DELETED — no longer in the table ✅
• C105: INSERTED — brand new customer added ✅

Step 7: Write the Updated Target Table Back to the Data Lake

# -------------------------------------------------------
# Overwrite the target location with the fully updated table
# -------------------------------------------------------
final_df.write.mode("overwrite").parquet(TARGET_PATH)

print("💾 Target table successfully overwritten with CDC-applied data.")

# -------------------------------------------------------
# Record the high-watermark for the next batch
# The next CDC extraction will use this timestamp as the starting point
# -------------------------------------------------------
max_watermark = cdc_deduped.agg(F.max("updated_at")).collect()[0][0]
print(f"🔖 High Watermark for next batch: {max_watermark}")

The high watermark is the maximum updated_at timestamp from this CDC batch. The next scheduled batch will use this value in its source query: SELECT * FROM customers WHERE updated_at > '{max_watermark}'.

Step 8: Verify — Read Back and Confirm

# -------------------------------------------------------
# Verification: Read the updated target and confirm correctness
# -------------------------------------------------------
verification_df = spark.read.parquet(TARGET_PATH)

print("🔎 Verification — Final Target Table Read-Back:")
verification_df.orderBy("customer_id").show(truncate=False)

# Assertions
assert verification_df.count() == 4, f"Expected 4 rows, got {verification_df.count()}"
assert verification_df.filter("customer_id = 'C104'").count() == 0, "C104 should be deleted"
assert verification_df.filter("customer_id = 'C105'").count() == 1, "C105 should be inserted"
assert verification_df.filter("customer_id = 'C101' AND city = 'Bangalore'").count() == 1, "C101 city should be Bangalore"

print("✅ All assertions passed! CDC pipeline verified successfully.")

The Complete Pipeline as a Single Reusable Function

Here is the entire CDC logic wrapped into a clean, production-ready function that can be called from Airflow, a cron job, or any scheduler:

def apply_cdc_batch(spark, target_path, cdc_events_df):
    """
    Apply a batch of CDC events (I/U/D) to an existing Parquet target table.

    Parameters:
        spark: SparkSession
        target_path: str - Path to the existing Parquet target table
        cdc_events_df: DataFrame - Must contain columns: customer_id, ..., updated_at, _cdc_operation
    """
    from pyspark.sql.window import Window

    # 1. Deduplicate — keep latest event per key
    w = Window.partitionBy("customer_id").orderBy(F.col("updated_at").desc())
    cdc_clean = cdc_events_df.withColumn("rn", F.row_number().over(w)) \
        .filter("rn = 1").drop("rn")

    # 2. Split by operation
    inserts = cdc_clean.filter("_cdc_operation = 'I'").drop("_cdc_operation")
    updates = cdc_clean.filter("_cdc_operation = 'U'").drop("_cdc_operation")
    deletes = cdc_clean.filter("_cdc_operation = 'D'").drop("_cdc_operation")

    # 3. Load existing target
    existing = spark.read.parquet(target_path)

    # 4. Remove affected keys from existing
    affected = updates.select("customer_id").union(deletes.select("customer_id"))
    surviving = existing.join(affected, on="customer_id", how="left_anti")

    # 5. Build new target = surviving + updates + inserts
    result = surviving.unionByName(updates).unionByName(inserts)

    # 6. Overwrite target
    result.write.mode("overwrite").parquet(target_path)

    # 7. Return watermark
    watermark = cdc_clean.agg(F.max("updated_at")).collect()[0][0]

    print(f"✅ CDC batch applied. Rows: {result.count()}, Next watermark: {watermark}")
    return watermark

Follow-Up Questions & Answers

Q1: What is the difference between CDC and ETL?

A: ETL (Extract-Transform-Load) is the overall pipeline pattern — you extract data, transform it, and load it somewhere. CDC is a specific extraction technique used within the "E" (Extract) phase of ETL. Instead of extracting the full table every run, CDC extracts only the changed rows. So CDC is a subset of ETL, not a replacement.

Q2: What happens if the source table doesn't have an updated_at column?

A: Without a timestamp column, timestamp-based CDC is impossible. Your alternatives are:

1. Log-based CDC (Debezium/DMS): Reads the database transaction log directly. No timestamp column needed.
2. Full snapshot comparison: Extract the entire source table every run and compare it row-by-row with the target to detect inserts, updates, and deletes. This works but is expensive at scale.
3. Add the column: Work with the application team to add updated_at with a database trigger or ORM default.

Q3: How do you handle CDC DELETEs when your target is append-only (e.g., S3 Parquet)?

A: Parquet on S3 does not support in-place row deletion. You have two options:

1. Full rewrite (used in our implementation): Read the entire target, filter out deleted keys, and overwrite the target location. Simple but expensive for very large tables.
2. Soft delete: Instead of removing the row, add an is_deleted = true flag and deleted_at timestamp. Downstream queries filter with WHERE is_deleted = false. Cheaper, but requires all consumers to respect the filter.

Q4: What is "at-least-once" delivery and why does deduplication matter?

A: Most CDC systems (Debezium + Kafka, AWS DMS) guarantee at-least-once delivery — meaning they guarantee every change event will be delivered, but the same event might be delivered more than once (e.g., during Kafka consumer rebalancing or network retries). Without deduplication, applying the same UPDATE event twice is harmless, but applying the same INSERT event twice would create duplicate rows. That's why Step 3 (deduplication using row_number() by key, ordered by timestamp descending) is critical.

Q5: How is log-based CDC different from timestamp-based CDC?

A:

Q6: Can CDC and SCD-2 be used together? How?

A: Absolutely — this is the standard production pattern. The pipeline flows like this:

1. Source DB  ──[CDC]──►  Raw/Staging Zone  (CDC captures the change events)
2. Staging    ──[SCD-2]──►  Dimension Table  (SCD-2 applies the history versioning)

• CDC extracts the change events and lands them in a raw staging area.
• SCD-2 logic then reads these events and applies the historical versioning (expire old rows, insert new versions with start_date, end_date, is_current).

The CDC file we built here is the first half of that pipeline. The SCD Type 2 scenario file covers the second half.

Sub-Scenarios

Sub-Scenario A: Handling Schema Evolution in CDC

Situation: The source database team adds a new column email to the customers table. CDC events now include the new field, but your existing Parquet target doesn't have it.

Fix: Use mergeSchema option when writing, or explicitly add the missing column:

# Add the new column with NULL to the existing target before union
existing_with_schema = existing_df.withColumn("email", F.lit(None).cast("string"))

# Now unionByName will succeed
result = existing_with_schema.unionByName(updates_df)

Sub-Scenario B: CDC with Multiple Tables (Multi-Table Sync)

Situation: You need to CDC-sync 50 tables from MySQL to S3. Writing 50 separate pipelines is unmaintainable.

Fix: Build a metadata-driven CDC framework:

# Define table metadata
tables = [
    {"source": "customers", "key": "customer_id", "target": "/lake/customers"},
    {"source": "orders",    "key": "order_id",    "target": "/lake/orders"},
    {"source": "products",  "key": "product_id",  "target": "/lake/products"},
]

# Generic CDC function — same logic, different table configs
for table in tables:
    cdc_events = spark.read.parquet(f"/staging/cdc/{table['source']}/")
    apply_cdc_batch(spark, table["target"], cdc_events)

Sub-Scenario C: Idempotent CDC — Re-running Failed Batches Safely

Situation: Your CDC pipeline crashed halfway through writing. If you re-run it, will you get duplicates?

Fix: The overwrite write mode makes our implementation naturally idempotent. Re-running the same batch produces the same output because:

1. We always read the full existing target.
2. We always reconstruct the entire result from scratch.
3. We always overwrite the entire target atomically.

For streaming CDC (Kafka), store the consumer offset checkpoint alongside the high watermark, and reset to the last committed offset on restart.