Change data capture: a practical Spark & Data Eng guide

Introduction

Database logs hold a treasure trove of business insights—every insert, update, and delete represents a moment in your data's lifecycle. Change data capture: a practical Spark & Data Eng guide shows you how to stream those database changes into your lakehouse in real time, keeping your analytics tables fresh and your decision-makers informed. We'll walk through the tools, patterns, and code you need to build a production CDC pipeline that handles millions of transactions without missing a beat.

Why CDC matters in your data stack

Traditional batch pipelines—running every night or every hour—miss the window between runs. If your e-commerce platform processes an order at 2:47 AM and your next batch job runs at 3:00 AM, you have a 13-minute blind spot. For compliance, fraud detection, or real-time dashboards, that lag is unacceptable. CDC captures changes as they happen, row by row, and streams them to your data warehouse or lakehouse. You get freshness without the cost of full table scans.

Beyond freshness, CDC enables audit trails and temporal queries. You can reconstruct data state at any point in time, critical for regulatory reporting and debugging production incidents.

Debezium reads database logs

At the heart of most CDC pipelines sits Debezium, an open-source distributed platform that reads transaction logs directly from your source databases. Instead of polling tables or running expensive queries, Debezium connects to PostgreSQL WAL (Write-Ahead Log), MySQL binlog, Oracle Archive Logs, or SQL Server transaction logs—whatever your primary database uses to ensure durability. It parses these logs, extracts semantic meaning, and emits structured change events.

Debezium runs as a Kafka Connect plugin. You configure a connector per source table or per entire schema. Here's a typical PostgreSQL connector setup:

# PostgreSQL Debezium Connector Configuration
connector_config = {
    "name": "postgres-orders-connector",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "database.hostname": "prod-db.internal",
        "database.port": "5432",
        "database.user": "debezium_user",
        "database.password": "secure_password",
        "database.dbname": "ecommerce",
        "database.server.name": "postgres-prod",
        "schema.include.list": "public",
        "table.include.list": "public.orders,public.items",
        "plugin.name": "pgoutput",
        "publication.name": "debezium_pub",
        "publication.autocreate.mode": "all_tables",
        "slot.name": "debezium_slot",
        "transforms": "route",
        "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
        "transforms.route.replacement": "$3"
    }
}

Debezium creates a replication slot in PostgreSQL and begins streaming the WAL. For each change, it emits a Kafka record with a before-image, after-image, and metadata (op=insert/update/delete, timestamp, LSN). The connector handles restarts gracefully: it checkpoints its position in the log, so if the connector crashes, it resumes from where it left off. No duplicates, no skipped changes.

In MySQL or Oracle, Debezium reads binlog or redo logs. The principle is identical: parse the database's native change stream, normalize it, and push to Kafka.

Publishes changes to Kafka

Each Debezium-captured change becomes a Kafka message. The topic name typically mirrors the source table (e.g., postgres-prod.public.orders). The message structure is standardized:

This format makes it easy for downstream consumers—your Spark jobs—to decide what to do. An update (op=u) has both before and after; you can compare them to find which columns changed. A delete (op=d) has only before; you know exactly what row was removed.

Kafka provides the buffer and replay capability. If your Spark job crashes or lags, Kafka retains the changes (configurable retention). Your consumer group tracks offset progress. This decoupling is critical: Debezium pushes changes independently of your lakehouse ingestion speed.

Typical message volume for a medium e-commerce platform: 1,000–10,000 changes per second during peak hours. Kafka scales horizontally; you partition topics by order ID or customer ID to maintain ordering per logical entity while spreading load.

Apply to lakehouse tables with MERGE

Now you have changes flowing into Kafka. The next step is consuming them in Spark and applying them to your lakehouse (Delta, Iceberg, or Hudi). The SQL MERGE operation is your tool here: it applies inserts, updates, and deletes in a single atomic transaction.

Here's a real example. You have a Delta table orders in production. Debezium streams changes to Kafka topic orders_cdc. Your Spark job:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, schema_of_json
import json

spark = SparkSession.builder.appName("CDC-Orders").enableDeltaLake().getOrCreate()

# Read CDC stream from Kafka
kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka-broker-1:9092,kafka-broker-2:9092") \
    .option("subscribe", "orders_cdc") \
    .option("startingOffsets", "latest") \
    .load()

# Parse CDC message
cdc_schema = """
{
    "before": {"order_id": "long", "customer_id": "long", "amount": "double", "status": "string"},
    "after": {"order_id": "long", "customer_id": "long", "amount": "double", "status": "string"},
    "op": "string",
    "source": {"ts_ms": "long"},
    "ts_ms": "long"
}
"""

parsed_df = kafka_df.select(
    from_json(col("value").cast("string"), cdc_schema).alias("cdc")
).select("cdc.*")

# Define merge logic
def merge_cdc(df, epoch_id):
    df.createOrReplaceTempView("cdc_changes")
    
    spark.sql("""
        MERGE INTO lakehouse.public.orders t
        USING (
            SELECT 
                after.order_id,
                after.customer_id,
                after.amount,
                after.status,
                op,
                ts_ms
            FROM cdc_changes
            WHERE op IN ('c', 'u', 'd')
        ) s
        ON t.order_id = s.order_id
        WHEN MATCHED AND s.op = 'd' THEN DELETE
        WHEN MATCHED AND s.op = 'u' THEN UPDATE SET
            t.customer_id = s.customer_id,
            t.amount = s.amount,
            t.status = s.status,
            t._cdc_ts = s.ts_ms
        WHEN NOT MATCHED AND s.op IN ('c', 'u') THEN INSERT
            (order_id, customer_id, amount, status, _cdc_ts)
        VALUES
            (s.order_id, s.customer_id, s.amount, s.status, s.ts_ms)
    """)

# Apply micro-batches
parsed_df.writeStream \
    .foreachBatch(merge_cdc) \
    .option("checkpointLocation", "/mnt/lakehouse/checkpoints/orders_merge") \
    .start() \
    .awaitTermination()

Each micro-batch, your Spark Streaming job pulls CDC records from Kafka, unpacks the JSON, and executes the MERGE. The op field drives the decision: delete for op='d', update for op='u', insert for op='c'. Delta handles the actual execution atomically—all changes in a batch either succeed or roll back as one unit. There are no partial states.

The lakehouse table now reflects the source database state with minimal lag, typically under a second for a healthy stream.

Handle deletes and schema changes

Deletes are straightforward in the MERGE: WHEN MATCHED AND s.op = 'd' THEN DELETE. But in practice, hard deletes are rare. Many teams use soft deletes: a deleted_at timestamp column. If your source table has a deleted_at column, Debezium captures it like any other update. Your MERGE clause then becomes:

spark.sql("""
    MERGE INTO lakehouse.public.orders t
    USING (
        SELECT 
            after.order_id,
            after.customer_id,
            after.amount,
            after.status,
            after.deleted_at,
            op,
            ts_ms
        FROM cdc_changes
    ) s
    ON t.order_id = s.order_id
    WHEN MATCHED THEN UPDATE SET
        t.customer_id = s.customer_id,
        t.amount = s.amount,
        t.status = s.status,
        t.deleted_at = s.deleted_at,
        t._cdc_ts = s.ts_ms
    WHEN NOT MATCHED THEN INSERT
        (order_id, customer_id, amount, status, deleted_at, _cdc_ts)
    VALUES
        (s.order_id, s.customer_id, s.amount, s.status, s.deleted_at, s.ts_ms)
""")

Then queries filter out soft-deleted rows naturally: WHERE deleted_at IS NULL.

Schema changes are trickier. If your source adds a new column, Debezium detects it and includes it in subsequent CDC messages. But your Spark MERGE references fixed columns. You need to handle schema evolution. Delta and Iceberg support column-level versioning, but the simplest approach is to use schema inference and dynamic column mapping:

# Enable automatic schema evolution in Delta
spark.sql("SET spark.databricks.delta.schema.autoMerge.enabled = true")

# Or use Iceberg with schema evolution
spark.conf.set("iceberg.schema-auto-migrate.enabled", "true")

# After adding columns to source, re-register the schema
parsed_df.writeStream \
    .option("mergeSchema", "true") \
    .foreachBatch(merge_cdc) \
    .option("checkpointLocation", "/mnt/lakehouse/checkpoints/orders_merge") \
    .start()

Alternatively, store raw CDC payloads (the full JSON from Kafka) in a staging table and apply schema inference at query time. This adds latency but avoids brittle upstream code. For a fast-moving schema (e.g., a growth-stage SaaS product), this is often the pragmatic choice.

Common mistakes

• Ignoring ordering guarantees: If two updates to the same order arrive out of order, your lakehouse ends up in an inconsistent state. Use Kafka partitioning by order ID to preserve per-entity causality. Debezium respects database transaction order within a partition.
• Not handling duplicates: Network hiccups can cause Spark to reprocess a micro-batch. Build idempotency into your MERGE by including a change log ID or LSN from Debezium's source metadata. Skip records you've already processed.
• Overflowing retention: Kafka retains data; Debezium produces unbounded streams. If your Spark job crashes for days, Kafka disk fills up. Set retention aggressively (e.g., 3 days) and monitor consumer lag.
• Missing initial snapshot: Debezium can stream a full table snapshot before log tailing. If you skip this, you miss historical data. Use "snapshot.mode": "initial" to capture baseline state before connecting to the binlog.
• Schema mismatch in MERGE: If you rename a source column but your MERGE still references the old name, inserts fail silently. Version your merge logic and test schema changes in staging first.

Key takeaways

• Debezium reads native database logs (WAL, binlog) and normalizes changes into structured Kafka topics—no polling, no full scans.
• Each CDC message contains before/after images and operation type (create, update, delete), giving your Spark job full context.
• Delta and Iceberg MERGE statements apply changes atomically, handling inserts, updates, and deletes in a single transaction.
• Soft deletes (marked by a timestamp column) are often safer and simpler than hard deletes; model them as data instead of operations.
• Schema evolution requires deliberate handling: either enable auto-migration in your lakehouse or stage raw payloads and infer schema at read time.
• Monitor Debezium connector health, Kafka consumer lag, and Spark job checkpoints; CDC's value vanishes if any component lags or fails silently.