Apply Change Data Capture with MERGE

You've built pipelines that land data, but your tables are still snapshots — every run overwrites yesterday's truth with today's. The moment multiple teams write to the same dataset, or upstream systems emit updates and deletes, a full reload doesn't just waste compute — it erases history and breaks downstream reports. Apply change data capture with MERGE solves this by letting you incrementally sync only what changed: new rows inserted, existing rows updated, stale rows deleted. In this lesson, you'll master the MERGE statement in Databricks to turn any Delta table into a living, auditable data asset — and you'll do it with a practical, hands-on exercise that you can run in your own workspace.

The problem this lesson solves

Imagine you're synchronizing a customer table from a transactional database into your Lakehouse. The source emits a nightly export with all rows, including ones that haven't changed. If you do a full overwrite, you:

• Waste compute — reprocessing millions of unchanged records.
• Break history — losing the ability to track when a record was first seen or last updated.
• Risk consistency — a failed overwrite can leave your table empty or partially loaded.

Even worse, what about deletions? A full export might not even include deleted rows, so you'd never know to remove them. This is the classic change data capture (CDC) problem: how do you efficiently apply a stream of changes — inserts, updates, deletes — to a target table without reprocessing everything?

The answer in Databricks is MERGE on Delta Lake. Delta tables support ACID transactions, time travel, and atomic operations, which make MERGE reliable and performant. With MERGE, you define exactly what happens when a source record matches an existing row (update), when it doesn't (insert), and optionally when no source matches (delete). This turns your pipeline from a brute-force reload into a surgical, incremental update.

Core concept / mental model

Think of MERGE as a three-way handshake between your existing target table, your incoming change data, and a set of rules you define.

• Target table: The Delta table that holds your current, authoritative state.
• Source table or DataFrame: A batch of changes (or the entire source dataset) you want to apply.
• Match condition: A key (like customer_id) that links the source and target.

When you run MERGE, Delta Lake performs the following for each source row:

• If a match is found in the target, apply the WHEN MATCHED THEN UPDATE action.
• If no match is found, apply the WHEN NOT MATCHED THEN INSERT action.
• Optionally, you can add WHEN NOT MATCHED BY SOURCE THEN DELETE to remove target rows that no longer exist in the source.

This is a mutating operation — it changes the target table in place, atomically. Unlike INSERT OVERWRITE, which replaces the whole table, MERGE touches only the rows that need to change, making it ideal for streaming or periodic CDC.

Here's a visual analogy: imagine you have a physical ledger (target). Every day, a courier brings you a list of changes (source). Instead of rewriting the entire ledger, you scan it, and for each entry you:

• Add a new line if the name is new (insert).
• Erase and rewrite the line if details changed (update).
• Cross out a line if the courier says the customer no longer exists (delete).

That's MERGE — a precise, auditable way to keep your ledger current.

How it works step by step

Let's break down the process of applying CDC with MERGE:

1. Prepare your source data: Ensure you have a table or DataFrame containing the changes. This could come from a streaming source (using readStream), a batch export, or a file in cloud storage.

2. Define the match key: Choose the column(s) that uniquely identify a record — for example, id or customer_id. This will be the join condition in the MERGE statement.

3. Write the MERGE statement: Use the MERGE INTO syntax. You'll specify: - The target table. - The source table or subquery. - The matching condition. - The actions for matched and unmatched rows.

4. Execute and verify: Run the statement, then check the target table to confirm the changes were applied correctly. You can also inspect the Delta table's history to see the transaction.

Key points to remember:

• The target table must be a Delta table. MERGE is not supported on non-Delta formats.
• The match condition must be deterministic and unambiguous — you can't match on columns that aren't unique, or you'll get an error.
• You can have multiple WHEN MATCHED clauses (e.g., update if a certain condition is true, otherwise ignore), but only the first matching clause is executed per row.

Hands-on walkthrough

Let's implement a complete CDC pipeline. We'll simulate a source table with changing customer data and a target Delta table, then apply the changes using MERGE.

Step 1: Set up the target table

In a Databricks notebook, create a Delta table to store customer data.

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

spark = SparkSession.builder.appName("CDC Merge Demo").getOrCreate()

# Define schema for customer data
schema = StructType([
    StructField("customer_id", IntegerType(), False),
    StructField("name", StringType(), True),
    StructField("email", StringType(), True),
    StructField("city", StringType(), True)
])

# Create some initial data
data = [
    (1, "Alice", "alice@example.com", "NYC"),
    (2, "Bob", "bob@example.com", "SF"),
    (3, "Carol", "carol@example.com", "LA")
]

# Create DataFrame and write to Delta table (overwrite for initial setup)
customers_df = spark.createDataFrame(data, schema)
customers_df.write.format("delta").mode("overwrite").save("/mnt/delta/customers")

# Register as a temp view for convenience
spark.sql("CREATE OR REPLACE TEMP VIEW customers_target USING delta OPTIONS (path '/mnt/delta/customers')")

Step 2: Generate change data

Now simulate an incoming CDC feed. It might include a new customer (4), an update to customer 2, and no information about customer 3 (which we'll later delete).

# Simulate CDC data from a source system
changes_data = [
    (2, "Robert", "robert@example.com", "Seattle"),  # update for id=2
    (4, "Diana", "diana@example.com", "Austin"),    # insert for id=4
]

changes_df = spark.createDataFrame(changes_data, schema)
changes_df.createOrReplaceTempView("changes")

Step 3: Apply MERGE

Run the MERGE statement to insert new records and update changed ones.

spark.sql("""
MERGE INTO customers_target AS target
USING changes AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
  UPDATE SET
    target.name = source.name,
    target.email = source.email,
    target.city = source.city
WHEN NOT MATCHED THEN
  INSERT (customer_id, name, email, city)
  VALUES (source.customer_id, source.name, source.email, source.city)
""")

# Verify the result
spark.sql("SELECT * FROM customers_target ORDER BY customer_id").show()

Expected output:

+-----------+------+-------------------+-------+
|customer_id| name | email             | city  |
+-----------+------+-------------------+-------+
|1          |Alice |alice@example.com  |NYC    |
|2          |Robert|robert@example.com|Seattle|
|3          |Carol |carol@example.com |LA     |
|4          |Diana |diana@example.com |Austin |
+-----------+------+-------------------+-------+

Step 4: Handle deletions with WHEN NOT MATCHED BY SOURCE

In many CDC scenarios, the source feed doesn't include the full dataset — it only sends changes. If you want to remove rows that no longer exist in the source (i.e., deleted upstream), you need to know the full source state. For this exercise, let's assume we have a full snapshot of the source and we want to remove any customer that isn't in that snapshot.

# Full source snapshot for this cycle — includes customers 1, 2, and 4 (but not 3)
full_source_data = [
    (1, "Alice", "alice@example.com", "NYC"),
    (2, "Robert", "robert@example.com", "Seattle"),
    (4, "Diana", "diana@example.com", "Austin")
]
full_source_df = spark.createDataFrame(full_source_data, schema)
full_source_df.createOrReplaceTempView("full_source")

spark.sql("""
MERGE INTO customers_target AS target
USING full_source AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
  UPDATE SET
    target.name = source.name,
    target.email = source.email,
    target.city = source.city
WHEN NOT MATCHED THEN
  INSERT (customer_id, name, email, city)
  VALUES (source.customer_id, source.name, source.email, source.city)
WHEN NOT MATCHED BY SOURCE THEN
  DELETE
""")

# After deletion, customer 3 should be gone
spark.sql("SELECT * FROM customers_target ORDER BY customer_id").show()

Expected output:

+-----------+------+-------------------+-------+
|customer_id| name | email             | city  |
+-----------+------+-------------------+-------+
|1          |Alice |alice@example.com  |NYC    |
|2          |Robert|robert@example.com|Seattle|
|4          |Diana |diana@example.com |Austin |
+-----------+------+-------------------+-------+

Pro tip: The WHEN NOT MATCHED BY SOURCE clause can be extremely powerful, but be careful — it deletes rows based on the entire source set. If your source only contains a partial batch (e.g., recent changes), you'll accidentally delete records not included in that batch. Use it only when the source truly represents the full target state.

Step 5: Apply CDC from a streaming source (bonus)

In real-world pipelines, changes often arrive as a stream. You can use foreachBatch combined with MERGE to apply micro-batches.

from pyspark.sql.functions import col

# Simulate a streaming DataFrame (in practice, you'd use readStream on Kafka or other sources)
streaming_changes = spark.readStream.format("rate").load() \
    .withColumn("customer_id", col("value") % 10) \
    .withColumn("name", col("value").cast("string")) \
    .withColumn("email", col("value").cast("string")) \
    .withColumn("city", col("value").cast("string")) \
    .select("customer_id", "name", "email", "city")

# This is a placeholder — real streaming MERGE would use foreachBatch
# streaming_changes.writeStream.foreachBatch(...).start()

In practice, you'd define a function that takes a micro-batch DataFrame and performs the MERGE. This pattern keeps your target table always up to date with near-zero latency.

Compare options / when to choose what

While MERGE is the star of CDC, it's not the only way to apply changes. Let's compare it with other common patterns.

When to choose MERGE:

• You have a key to join on and need to support updates and deletes.
• You want incremental processing to save cost and time.
• You rely on Delta's ACID guarantees and want a single atomic operation.

When to avoid MERGE:

• Your workload is purely append-only (e.g., log events) — a simple INSERT is cheaper.
• Your source is huge and full-replace is acceptable — but even then, MERGE can help with schema drift.
• Your target isn't Delta — you'd need to convert it first.

Pro tip: If you're doing frequent small MERGEs, consider optimizing your table with OPTIMIZE and ZORDER BY on the join key to speed up the matching step. MERGE performance depends on finding the right files.

Troubleshooting & edge cases

Even experienced engineers hit snags with MERGE. Here are common issues and how to fix them.

Error: The number of conditions in the WHEN MATCHED clause must be equal to the number of conditions in the WHEN NOT MATCHED clause

This happens when you have multiple WHEN MATCHED clauses but only one WHEN NOT MATCHED, or vice versa. Spark requires that the number of clauses is consistent — actually, you can have multiple MATCHED and one NOT MATCHED, and it's allowed. But if you use WHEN NOT MATCHED BY SOURCE, you need to ensure your WHEN clauses are structured correctly. The exact error message may vary; the key is to check that you haven't missed a condition.

Solution: Review your MERGE syntax—ensure that for each WHEN MATCHED with a condition, you have a corresponding WHEN NOT MATCHED (if needed) and that WHEN NOT MATCHED BY SOURCE is only used when appropriate.

Error: MERGE destination only supports Delta tables

MERGE only works on Delta Lake. If you're trying to merge into a Parquet or CSV table, you'll get this error.

Solution: Convert your table to Delta (CONVERT TO DELTA) or rewrite it as a Delta table first.

Issue: Updates don't appear, or deleted rows remain

You likely forgot to include the WHEN NOT MATCHED BY SOURCE clause, or your match condition is too restrictive (e.g., comparing multiple columns that don't all match).

Solution: Print the source and target to verify the key values. Ensure the join condition is simple and correct: ON target.id = source.id.

Issue: Slow MERGE on large tables

If your table has billions of rows and you're matching on a non-optimized column, the MERGE can scan too many files.

Solution: Use OPTIMIZE and ZORDER BY on the join key:

OPTIMIZE customers ZORDER BY (customer_id);

Also, try to filter the source to only include changed rows — don't push the entire source table if you can avoid it.

Edge case: Duplicate keys in source

If your source data has multiple rows with the same join key, MERGE will fail with a Duplicate keys error. This happens because MERGE can't decide which row to apply.

Solution: Deduplicate the source before merging:

changes_df = changes_df.dropDuplicates(["customer_id"])

Edge case: Schema drift

If the source schema evolves (new columns appear), MERGE may fail or ignore the new columns. Delta can handle schema evolution with mergeSchema option:

changes_df.write.option("mergeSchema", "true").format("delta").mode("append").save("/mnt/delta/customers")

But for MERGE itself, you need to explicitly handle new columns in the UPDATE SET clause.

What you learned & what's next

You've now mastered apply change data capture with MERGE — you understand the pain of full reloads, the mental model of a three-way handshake, and how to implement it step by step in Databricks. You can:

• Explain the core idea behind CDC and why MERGE is the right tool.
• Complete a practical exercise using MERGE INTO with inserts, updates, and deletes.
• Apply MERGE in a streaming context with foreachBatch.
• Troubleshoot common MERGE issues like slow performance, duplicate keys, and schema drift.

This skill is foundational for building reliable, incremental ETL pipelines. Next in your Databricks journey, you'll likely explore Delta Live Tables to orchestrate these merges declaratively, or change data feed to capture changes out of your Delta tables. You're now ready to handle real-world CDC scenarios with confidence.

Pro tip: Always test your MERGE logic on a small, sample dataset before running it in production. Use DESCRIBE HISTORY to audit what changed in each transaction — this gives you a full audit trail of your data pipeline.