DEV Community: Karan Pulyala

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

Karan Pulyala — Tue, 20 Oct 2026 14:20:57 +0000

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:

Source row	`merge_key`	Result
Brand-new customer	its id	No match, inserted as current
Existing, unchanged	its id	Matches, but condition is false, so nothing happens
Existing, changed (copy 1)	its id	Matches, old row closed (`is_current = false`)
Existing, changed (copy 2)	NULL	No match, new current row inserted

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.

Diagnosing and Fixing Data Skew in PySpark on Azure Databricks (2026 Edition)

Karan Pulyala — Mon, 19 Oct 2026 20:11:31 +0000

A familiar failure: the stage hits 99% and stalls. 199 tasks are done, one is still running, and adding nodes doesn't help, because the bottleneck is one partition, not the cluster.

That's data skew. In this post I'll go through how I diagnose it and how I fix it on Azure Databricks as of 2026, roughly in the order I'd try each fix. Every snippet runs on Databricks Free Edition, which is serverless-only, so I'll flag the places where serverless behaves differently.

Not affiliated with or endorsed by Databricks. Config names and defaults are from the Apache Spark and Databricks docs as of October 2026. Check them against your runtime version.

What skew actually is

A shuffle sends every row with the same key to the same partition. That's what makes joins and groupBy work. If one key holds 30% of your rows, one task gets 30% of the work, and the stage waits for it.

The usual culprits:

NULL or placeholder keys ('UNKNOWN', -1, ''). In my experience this is the cause more often than anything else.
A handful of "whale" entities: one huge customer, one default store, one bot user agent.
Time skew: one day (Black Friday, a backfill date) is 50Ã— the others.
File/read skew, which isn't about keys at all. A few huge or unsplittable files (gzipped CSV, for example) make the first stage lopsided.

Step 1: Build a skewed dataset to practice on

Paste this into a Free Edition notebook. It creates 20M orders: ~30% with a NULL customer, ~20% for one whale customer (42):

from pyspark.sql import functions as F

N = 20_000_000
orders = (
    spark.range(N)
    .withColumn("r", F.rand(seed=7))
    .withColumn(
        "customer_id",
        F.when(F.col("r") < 0.30, F.lit(None).cast("bigint"))   # ~30% NULL
         .when(F.col("r") < 0.50, F.lit(42).cast("bigint"))     # ~20% whale
         .otherwise((F.rand(seed=11) * 1_000_000).cast("bigint"))
    )
    .withColumn("amount", F.round(F.rand(seed=3) * 100, 2))
    .drop("r")
)

customers = (
    spark.range(1_000_000).withColumnRenamed("id", "customer_id")
    .withColumn("segment", F.when(F.col("customer_id") % 10 == 0, "enterprise").otherwise("smb"))
)

Step 2: Confirm it. Don't guess.

Key distribution. This is the cheapest check and the one I always run first:

total = orders.count()
(orders.groupBy("customer_id").count()
    .withColumn("pct_of_rows", F.round(100 * F.col("count") / total, 2))
    .orderBy(F.desc("count"))
    .show(10))

Double-digit percentages at the top mean you've found it.

Partition-level distribution. This shows what a shuffle on that key does to your tasks:

parts = (orders.repartition(200, "customer_id")
    .groupBy(F.spark_partition_id().alias("pid")).count())

parts.agg(
    F.max("count").alias("max_rows"),
    F.percentile_approx("count", 0.5).alias("median_rows"),
).withColumn("max_over_median", F.col("max_rows") / F.col("median_rows")).show()

A max/median ratio in the tens or hundreds means one task will do most of the work, no matter how big your cluster is.

In the Spark UI (classic compute). Open the slow stage and look at Summary Metrics. Compare Max with Median for Duration, Shuffle Read Size / Records, and Spill (Disk). A healthy stage has a fairly flat distribution. A skewed stage has a median of seconds and a max of minutes, and the Event Timeline shows one long bar.

On serverless (including Free Edition) there's no Spark UI. Click See performance under the cell and open the query profile. Look for the operator with a disproportionate share of time and rows. It's operator-level, not a per-task histogram, so the key-distribution query above is even more useful there.

Step 3: Let AQE do its job (and know when it can't)

Adaptive Query Execution re-plans at stage boundaries using real shuffle statistics. It's on by default in Spark 3.2+ and on Databricks. Its skew-join feature splits an oversized partition into smaller ones and replicates the matching partition on the other side. A partition counts as skewed only when both of these are true:

it's larger than spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5) Ã— the median partition size, and
it's larger than spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (default 256MB).

Split sizes aim at spark.sql.adaptive.advisoryPartitionSizeInBytes (default 64MB). When it works, the final plan shows a SortMergeJoin with isSkew=true.

joined = orders.join(customers, "customer_id", "left")
joined.write.mode("overwrite").format("noop").save()   # run it without writing anything

noop runs the whole job without writing output. Then check the plan in the query profile (serverless) or the SQL tab (classic).

What AQE won't handle, according to the docs:

Broadcast joins. Skew handling only applies to sort-merge and shuffle-hash joins.
Some join sides. In a LEFT OUTER join only the left side can be split. FULL OUTER gets no skew handling at all.
Partitions under 256MB. If your "skew" is 200MB against a 2MB median, the threshold blocks it. On classic compute you can lower the thresholds. On serverless you can't, because only a short allowlist of Spark confs (like spark.sql.shuffle.partitions, which defaults to auto there) can be set.
Scan-stage skew, aggregations and window functions. The skew-join rule is about joins.

One honest caveat about the demo: on a small Free Edition dataset the skewed partition may not cross 256MB, so you may not see isSkew=true. That's the threshold working as designed, not a broken demo.

Step 4: Broadcast the small side

If the other side fits in memory, a broadcast hash join removes the shuffle completely. With no shuffle there are no hot partitions, so skew stops mattering:

joined = orders.join(F.broadcast(customers), "customer_id", "left")

Spark auto-broadcasts below spark.sql.autoBroadcastJoinThreshold (10MB in OSS). Databricks AQE can also switch to a broadcast at runtime below spark.databricks.adaptive.autoBroadcastJoinThreshold (documented at 30MB). Databricks' own guidance is still to hint the broadcast when you know a side is small, because a planned broadcast beats one that AQE only chooses after both sides have already shuffled.

Two cautions: compressed Parquet can expand a lot in memory, and you can't broadcast the preserved side of an outer join (the left side of a LEFT JOIN).

Step 5: Treat hot keys separately (especially NULLs)

For NULL keys this is free. NULLs never match in an equi-join, so pushing them through the shuffle is pure waste:

is_null = F.col("customer_id").isNull()

matched = orders.filter(~is_null).join(customers, "customer_id", "left")
nulls   = orders.filter(is_null)            # no partner rows to find anyway

result = matched.unionByName(nulls, allowMissingColumns=True)

For an inner join, just filter the NULLs out. The same split works for a known whale key: send it through a broadcast path and the rest through the normal join.

Step 6: Salting, the targeted version

Salting spreads one key across N partitions by adding a random suffix on the big side and replicating the small side N times. The common version salts every key, which multiplies the whole dimension table by N. I prefer to salt only the hot keys:

SALT = 16
hot_keys = [42]  # take these from your Step 2 profiling, not from a hard-coded guess

orders_s = orders.withColumn(
    "salt",
    F.when(F.col("customer_id").isin(hot_keys), (F.rand(seed=1) * SALT).cast("int"))
     .otherwise(F.lit(0))
)

customers_s = customers.withColumn(
    "salt",
    F.explode(
        F.when(F.col("customer_id").isin(hot_keys), F.sequence(F.lit(0), F.lit(SALT - 1)))
         .otherwise(F.array(F.lit(0)))
    )
)

salted = orders_s.join(customers_s, ["customer_id", "salt"], "left").drop("salt")

Only the whale's dimension row gets replicated (16 rows, not 16 million). Size SALT so each slice of the hot key is close to a normal partition.

For aggregations, be selective too. Spark already does partial aggregation on the map side, so plain sum/count on a skewed key often hurts less than you'd expect. Salting pays off for aggregations that can't be combined that way, like collect_list, exact countDistinct and percentiles, and for window functions partitioned by a hot key. Two-phase aggregation looks like this:

partial = (orders.withColumn("salt", (F.rand(seed=5) * SALT).cast("int"))
    .groupBy("customer_id", "salt")
    .agg(F.sum("amount").alias("s"), F.count("*").alias("c")))

final = partial.groupBy("customer_id").agg(F.sum("s").alias("total_amount"),
                                           F.sum("c").alias("order_count"))

Step 7: Repartitioning (and its trap)

repartition(400, "customer_id") does not fix key skew. Every row for key 42 still hashes to the same partition, and you've paid for a shuffle. What helps:

repartition(400) with no key (round-robin) when you just need even work for narrow, row-by-row processing.
The REBALANCE hint before a write. AQE can split skewed partitions in a rebalance (spark.sql.adaptive.optimizeSkewsInRebalancePartitions.enabled, default true), which avoids one giant output file:

orders.hint("rebalance").write.mode("overwrite").saveAsTable("workspace.default.orders_demo")

(workspace.default is the default catalog and schema on Free Edition. Use your own elsewhere.)

Step 8: Fix the layout when the skew is on read

If the first stage is the lopsided one, AQE can't help, because there's no shuffle statistic to re-plan from yet. That's a file-layout problem: a few giant files, thousands of tiny ones, unsplittable compression, or Hive-style partitions where one date= folder is huge.

On Delta, the current recommendation for new tables is liquid clustering. Databricks lists "tables with heavy data skew" among its target use cases:

spark.sql("""
  CREATE OR REPLACE TABLE workspace.default.orders_clustered
  CLUSTER BY (customer_id)
  AS SELECT * FROM workspace.default.orders_demo
""")
spark.sql("OPTIMIZE workspace.default.orders_clustered")

Be clear about what this buys you: evenly sized files and better data skipping. It does not remove skew from a later join on customer_id. The shuffle still sends key 42 to one partition. With predictive optimization on Unity Catalog managed tables, CLUSTER BY AUTO lets Databricks choose the keys.

How to explain this in an interview

Interviewers are checking whether you have a process, not a definition. A structure that works:

Symptom: "A stage stuck at 99% with one long task. Max task duration far above the median, a big shuffle read on that task, often spill."
Confirm: "I profile the key distribution, and the first thing I check is NULL and default keys."
Fix in order of cost: "Check that AQE skew join is applicable (join type, thresholds), broadcast the small side if I can, route NULL or hot keys around the join, and use targeted salting as a last resort."
Know the limits: "AQE only handles skew in shuffle joins. It won't touch broadcast joins or scan-stage skew, and on serverless I can't tune its thresholds."
Close the loop: "Then I verify. I check for isSkew=true in the plan, compare max against median again, and add a skew check to the pipeline so a new whale customer doesn't bring it back."

Point 5 is where you stand out. Most candidates stop at "I'd use salting."

Checklist

[ ] Max vs. median task duration and shuffle read (Spark UI), or the slowest operator (query profile)
[ ] Top-10 key distribution, including NULL and placeholder values
[ ] Is it a join, an aggregation, a window function, or the scan stage?
[ ] AQE on and applicable: shuffle join, a join type that allows splitting that side, partitions over the thresholds
[ ] Small side broadcastable? Hint it.
[ ] NULLs filtered or routed around the join
[ ] Salt only the hot keys, sized from real numbers
[ ] Scan-stage skew â†’ fix the files (compaction, liquid clustering), not the join
[ ] Re-measure, then add monitoring

If you're prepping for Databricks interviews, I put 10 questions like this with model answers into a free sample: https://pulyala.gumroad.com/l/vonemy (the full 61-question pack is here: https://pulyala.gumroad.com/l/ivjsaf).

Not affiliated with Databricks. Databricks and Azure Databricks are trademarks of their respective owners.