Definition
Delta Lake's MERGE INTO is the core tool for atomic upserts.
- Atomic transaction – The entire match-and-apply executes as one Delta commit: it either fully applies or not at all.
- WHEN MATCHED – Updates or deletes existing rows that match the join condition.
- WHEN NOT MATCHED – Inserts new rows from the source that have no match in the target.
- Concurrency safe – Unlike a manual read-check-write, MERGE uses optimistic concurrency control and detects conflicting commits.
- Efficient – Rewrites only the files containing matched rows, not the whole table.
This replaces the error-prone pattern of reading the target, computing differences, and writing back.
Imagine the National Bank of India gets a nightly file of balance changes. For each account, if it already exists, update the balance; if it is new, insert it; if it is closed, delete it. Doing this by hand means reading each account, checking, then writing, and if two clerks do this at once they can overwrite each other. MERGE INTO is one clerk who does the whole match-and-apply in a single locked step, so the result is always correct.
Understanding the Concepts
What is MERGE INTO?
MERGE INTO is Delta Lake's atomic upsert operation. It joins a target table with a source dataset on a matching condition and, in a single transaction, applies WHEN MATCHED actions (UPDATE or DELETE) and WHEN NOT MATCHED actions (INSERT).
-- Basic MERGE for the National Bank of India nightly balance changes
MERGE INTO banking_prod.accounts AS target
USING nightly_changes AS source
ON target.account_id = source.account_id
WHEN MATCHED AND source.status = 'CLOSED' THEN DELETE
WHEN MATCHED THEN UPDATE SET
target.balance = source.balance,
target.last_updated = source.last_updated
WHEN NOT MATCHED THEN INSERT (
account_id, customer_name, balance, branch_code, status, last_updated
) VALUES (
source.account_id, source.customer_name, source.balance,
source.branch_code, source.status, source.last_updated
);
How does the DataFrame API work?
In PySpark, you use the DeltaTable class to express the same MERGE programmatically.
from delta.tables import DeltaTable
from pyspark.sql import functions as F
target = DeltaTable.forPath(spark, "/data/bank/accounts")
source = spark.read.parquet("/data/bank/nightly_changes")
target.alias("t").merge(
source.alias("s"),
"t.account_id = s.account_id"
).whenMatchedUpdate(
condition="s.status = 'CLOSED'",
set={"t.status": "'CLOSED'", "t.balance": "0"}
).whenMatchedUpdate(
set={
"t.balance": "s.balance",
"t.last_updated": "s.last_updated"
}
).whenNotMatchedInsert(
values={
"t.account_id": "s.account_id",
"t.customer_name": "s.customer_name",
"t.balance": "s.balance",
"t.branch_code": "s.branch_code",
"t.status": "s.status",
"t.last_updated": "s.last_updated"
}
).execute()
Why is MERGE better than a manual read-check-write?
The manual approach reads the target, computes which rows to update or insert, and writes them back. Between the read and the write, another writer can change the table, so your write is based on stale state.
# WRONG: manual read-check-write (race condition)
target_df = spark.read.format("delta").load("/data/bank/accounts")
source_df = spark.read.parquet("/data/bank/nightly_changes")
# Compute updates and inserts
updates = target_df.join(source_df, "account_id", "inner")
new_rows = source_df.join(target_df, "account_id", "left_anti")
# Between this read and the write below, another job may change the table
# Your write is based on STALE state -- you can lose updates or create duplicates
updates.write.format("delta").mode("overwrite").save("/data/bank/accounts")
new_rows.write.format("delta").mode("append").save("/data/bank/accounts")
# RIGHT: use MERGE instead (atomic, no race)
MERGE avoids this because Delta performs the join and the writes inside one optimistic transaction, detecting conflicting commits and failing or retrying rather than silently corrupting data.
How does MERGE handle concurrency?
Delta uses optimistic concurrency control. MERGE reads the target files, computes the changes, and attempts to commit. If another writer modified the same files between the read and the commit, Delta detects the conflict and either retries or throws a ConcurrentModificationException.
Job A: MERGE starts at version 5, reads files [f1, f2] Job B: MERGE starts at version 5, reads files [f1, f3] Job A commits version 6: rewrites f1 → f1' Job B tries to commit version 7: detects f1 was changed → Conflict! Delta retries or raises ConcurrentModificationException Result: no silent corruption, no lost updates
How does MERGE handle multiple changes per key in one batch?
If a batch contains multiple changes for the same key, the source must not match a target row more than once. Delta rejects ambiguous matches. You must first reduce to the latest change per key.
from pyspark.sql.window import Window
# Multiple changes for the same account in one batch
raw_changes = spark.read.parquet("/data/bank/nightly_changes")
# Reduce to the latest change per account using sequence number
w = Window.partitionBy("account_id").orderBy(F.col("sequence").desc())
latest_changes = raw_changes.withColumn("rn", F.row_number().over(w)) \
.filter("rn = 1").drop("rn")
# Now each account appears at most once -- safe for MERGE
target = DeltaTable.forPath(spark, "/data/bank/accounts")
target.alias("t").merge(
latest_changes.alias("s"),
"t.account_id = s.account_id"
).whenMatchedUpdateAll() \
.whenNotMatchedInsertAll() \
.execute()
How do you optimize MERGE performance?
MERGE rewrites only the files that contain matched rows. You can make it faster by narrowing the match condition with partition or ZORDER predicates.
-- Narrow the merge to only touch the relevant partition
MERGE INTO banking_prod.accounts AS target
USING nightly_changes AS source
ON target.account_id = source.account_id
AND target.branch_code = source.branch_code -- partition predicate
WHEN MATCHED THEN UPDATE SET target.balance = source.balance
WHEN NOT MATCHED THEN INSERT *;
-- ZORDER the table on the merge key for better file pruning
OPTIMIZE banking_prod.accounts ZORDER BY (account_id);
What is whenNotMatchedBySourceDelete?
For full-sync upserts where you want to remove target rows absent from the source, use WHEN NOT MATCHED BY SOURCE THEN DELETE.
-- Full sync: delete accounts that no longer exist in the source
MERGE INTO banking_prod.accounts AS target
USING source_system.accounts AS source
ON target.account_id = source.account_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
WHEN NOT MATCHED BY SOURCE THEN DELETE;
-- Target now exactly mirrors the source
What should a senior engineer know?
- MERGE is the standard pattern for CDC (Change Data Capture) in Delta Lake.
- Delta rejects a MERGE where the source matches a target row more than once. Always deduplicate the source first.
- MERGE can be expensive if the match condition is too broad, touching many files. Partition pruning and ZORDER help.
- For very large tables, consider partitioning by the merge key's prefix or date to limit file rewrites.
- MERGE is idempotent when the source is deterministic: running the same merge twice produces the same result.
# Verify merge results
accounts = spark.read.format("delta").load("/data/bank/accounts")
print(f"Total accounts: {accounts.count()}")
print(f"Closed accounts: {accounts.filter('status = \"CLOSED\"').count()}")
Visual Overview
MERGE INTO Flow:
Source (nightly changes) Target (accounts table)
┌─────────────────────┐ ┌─────────────────────┐
│ ACC-1001 50000 UPD │ │ ACC-1001 45000 │
│ ACC-2045 CLOSED DEL │ │ ACC-2045 120000 │
│ ACC-9999 30000 NEW │ │ ACC-3000 80000 │
└──────────┬──────────┘ └──────────┬──────────┘
│ │
└───────────┬───────────────┘
▼
┌─────────────────────┐
│ MERGE (atomic) │
│ ON account_id │
│ │
│ MATCHED + UPD → │
│ UPDATE balance │
│ MATCHED + DEL → │
│ DELETE row │
│ NOT MATCHED → │
│ INSERT row │
└──────────┬──────────┘
▼
┌─────────────────────┐
│ Result (one commit) │
│ ACC-1001 50000 │
│ ACC-3000 80000 │
│ ACC-9999 30000 │
└─────────────────────┘
Key Takeaway
MERGE INTO is Delta Lake's atomic upsert. You give it a target table, a source dataset, and an ON condition that defines what counts as a match, usually a business key like account_id. Then you specify branches: WHEN MATCHED can UPDATE or DELETE, and WHEN NOT MATCHED can INSERT. The whole thing commits as one Delta transaction, so it is atomic and isolated, and it rewrites only the files that contain matched rows.
The contrast with a manual read-check-write is the key interview point. In the manual approach, between your read and your write another job can change the table, so your write is based on stale state and you either lose updates or create duplicates. MERGE avoids that because Delta performs the join and the writes inside a single optimistic transaction and detects conflicting commits.
At the National Bank of India we used MERGE for the nightly balance change file. Each night some accounts had new balances, some were new, and some were closed, and we applied all three cases in one statement. Because it was a single atomic commit, a report running at the same time always saw a consistent snapshot. The caution is to keep the match condition narrow with partition or ZORDER predicates and to ensure the source never matches a target row more than once.
Remember one sentence: MERGE INTO atomically matches, updates, deletes, and inserts in one transaction, replacing the race-prone read-check-write pattern with a safe, efficient upsert.
---