Delta Lake MERGE in PySpark: Upserts, SCD Type 2, and the Traps Interviewers Ask About

MERGE INTO is the workhorse of almost every Delta Lake pipeline on Azure Databricks. It's also one of the most common follow-up topics in data engineering interviews: once you say "I upsert into Delta," the next question is usually "OK, what happens when the source has duplicates?" or "How would you build SCD Type 2 with it?"

This post walks through the patterns I'd want a candidate (or a teammate) to know cold, with PySpark code you can run in a notebook.

1. The basic upsert

from delta.tables import DeltaTable
from pyspark.sql import functions as F

target = DeltaTable.forName(spark, "silver.customers")
updates = spark.table("bronze.customer_updates")

(target.alias("t")
    .merge(updates.alias("s"), "t.customer_id = s.customer_id")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute())

Delta joins source to target on the merge condition, finds the files that contain matched rows, rewrites them, and commits new files atomically. Readers never see a half-applied merge.

The SQL equivalent:

MERGE INTO silver.customers t
USING bronze.customer_updates s
ON t.customer_id = s.customer_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

2. Trap #1: duplicate keys in the source

If two source rows match the same target row, the merge fails with an error along the lines of "multiple source rows matched and attempted to modify the same target row." Delta refuses to guess which one wins, which is the right call.

Interviewers love this one because the fix shows whether you think about data quality. Deduplicate the source first, keeping the latest record per key:

from pyspark.sql.window import Window

w = Window.partitionBy("customer_id").orderBy(F.col("updated_at").desc())

latest_updates = (updates
    .withColumn("rn", F.row_number().over(w))
    .filter("rn = 1")
    .drop("rn"))

Use row_number(), not rank(), so ties on updated_at still produce exactly one row. If ties are realistic, add a second ordering column (an ingestion timestamp or file name) so the result is deterministic.

3. Trap #2: rewriting rows that didn't change

whenMatchedUpdateAll() with no condition updates every matched row, even when nothing changed. In a daily full-snapshot feed where 98% of rows are identical, that means rewriting a lot of files for no reason.

Add a condition so only real changes are written. Use the null-safe equality operator <=>, because a <> b is NULL (not true) when either side is NULL:

change_cond = "NOT (t.email <=> s.email AND t.city <=> s.city AND t.tier <=> s.tier)"

(target.alias("t")
    .merge(latest_updates.alias("s"), "t.customer_id = s.customer_id")
    .whenMatchedUpdateAll(condition=change_cond)
    .whenNotMatchedInsertAll()
    .execute())

For wide tables, comparing a hash of the business columns is easier to maintain:

cols = ["email", "city", "tier"]
hashed = latest_updates.withColumn(
    "row_hash", F.sha2(F.concat_ws("||", *[F.coalesce(F.col(c).cast("string"), F.lit("<null>")) for c in cols]), 256)
)
# store row_hash in the target too, then: condition="t.row_hash <> s.row_hash"

4. Trap #3: scanning the whole target

The merge condition decides how much of the target Delta has to read. If the target is partitioned by event_date and your source only touches the last few days, say so in the condition:

(target.alias("t")
    .merge(
        latest_updates.alias("s"),
        "t.event_date >= date_sub(current_date(), 3) AND t.event_id = s.event_id AND t.event_date = s.event_date"
    )
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute())

A literal-style predicate on the partition (or clustering / Z-order) column lets Delta prune files before the join. Without it, every merge reads the full table, and your job gets slower every day as the table grows. This is one of the most common "why is my merge getting slower?" root causes I've seen.

Two related knobs worth mentioning in an interview:

• If the source is small, Spark will usually broadcast it. Check the plan in the Spark UI rather than assuming.
• On newer Databricks runtimes, deletion vectors let Delta mark rows as deleted/updated without rewriting whole files immediately, which can make merges with few changed rows much cheaper. Run OPTIMIZE regularly so the table doesn't accumulate small files.

5. Handling deletes from the source system

If the source is a full snapshot and a key disappears, you may want to delete (or soft-delete) it in the target. Delta Lake 2.3+ / recent Databricks runtimes support whenNotMatchedBySource:

(target.alias("t")
    .merge(latest_updates.alias("s"), "t.customer_id = s.customer_id")
    .whenMatchedUpdateAll(condition=change_cond)
    .whenNotMatchedInsertAll()
    .whenNotMatchedBySourceUpdate(set={"is_active": F.lit(False)})
    .execute())

Only do this when the source really is a complete snapshot. Running it against an incremental feed will "deactivate" every row that just didn't change today.

6. SCD Type 2 with a single MERGE

The classic question: keep full history, with is_current, start_date, and end_date columns. A changed customer needs two actions in one atomic commit: close the old row and insert a new current row. A single MERGE can't both update and insert for the same source row, so the trick is to stage the changed rows twice.

dim = DeltaTable.forName(spark, "gold.dim_customer")
src = latest_updates  # customer_id, email, city, tier, updated_at

current = dim.toDF().filter("is_current = true")

scd_change = "NOT (t.email <=> s.email AND t.city <=> s.city AND t.tier <=> s.tier)"

# Rows whose attributes changed: these need a NEW current row inserted
changed = (src.alias("s")
    .join(current.alias("t"), "customer_id")
    .where(scd_change)
    .select("s.*"))

staged = (
    # merge_key = NULL never matches, so these rows fall through to INSERT
    changed.withColumn("merge_key", F.lit(None).cast(src.schema["customer_id"].dataType))
    .unionByName(
        # merge_key = customer_id matches the current row, so we can close it out
        src.withColumn("merge_key", F.col("customer_id"))
    )
)

(dim.alias("t")
    .merge(staged.alias("s"), "t.customer_id = s.merge_key AND t.is_current = true")
    .whenMatchedUpdate(
        condition=scd_change,
        set={"is_current": F.lit(False), "end_date": F.col("s.updated_at")}
    )
    .whenNotMatchedInsert(values={
        "customer_id": "s.customer_id",
        "email": "s.email",
        "city": "s.city",
        "tier": "s.tier",
        "is_current": F.lit(True),
        "start_date": F.col("s.updated_at"),
        "end_date": F.lit(None).cast("timestamp"),
    })
    .execute())

What happens to each kind of source row:

Because it's one MERGE, the close-out and the insert land in the same Delta commit, so no reader ever sees a customer with zero or two current rows.

7. Verify what the merge actually did

Don't guess, check. Every merge records operation metrics in the table history:

(spark.sql("DESCRIBE HISTORY gold.dim_customer LIMIT 1")
    .select("version", "operation", "operationMetrics")
    .show(truncate=False))

Look at numTargetRowsUpdated, numTargetRowsInserted, numTargetFilesAdded, and numTargetFilesRemoved. If a merge that changed 200 rows removed 4,000 files, your condition isn't pruning and that's where to dig.

Quick interview checklist

• Deduplicate the source before merging, and explain why Delta errors instead of picking a row.
• Use null-safe comparisons (<=>) in change detection.
• Add a condition to whenMatched so unchanged rows aren't rewritten.
• Put partition or clustering predicates in the merge condition so files get pruned.
• Know the staged-union trick for SCD Type 2 in a single atomic merge.
• Read DESCRIBE HISTORY metrics to prove what happened.

If you're prepping for a Databricks or PySpark interview, I put together a free 10-question sample with worked answers in this style. The full Azure Databricks & PySpark Interview Pack has the rest if you find it useful.