A practical PySpark contract for duplicate events, conflicting versions, and deterministic updates.
Three source rows. One target row. Everyone insists they are the right match. Spider-Man pointing meme; image source: UOL / reproduction.
Three records walk into a MERGE. All claim to represent the same customer. One carries yesterday’s address. Two carry today’s address. The target table has one row, and Delta can refuse to update it from more than one matching source row, even when those rows agree.
If you have seen DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE, this article is for you.
The engineering problem is less about persuading Delta to accept the input and more about defining what the input means. Which records describe the same entity? What makes a version newer? When is a duplicate harmless, and when is it evidence that the producer has contradicted itself?
Those decisions belong in a data contract. A successful job is a poor substitute for one.
First, decide what “the same customer” means
Consider this synthetic batch:
key | sequence | value
" a " | 1 | old
"A" | 2 | new
"A" | 2 | new
For this example, the business considers letter case and leading or trailing spaces irrelevant. We therefore normalize the key with upper(trim(key)). All three records identify A.
This is a domain assumption, not a universal cleaning rule. If abc and ABC are different product codes in your system, converting them to uppercase would erase a meaningful distinction. Also, Spark's ordinary trim is not a complete Unicode normalization strategy. Be precise about the spaces and characters your contract accepts.
Apply the same identity rule everywhere it matters: validation, aggregation, deduplication, and the eventual match condition. Grouping original keys and normalizing only inside ON can produce several supposedly unique groups that collapse into one match key.
The records did not become duplicates because MERGE was feeling dramatic. The transformation exposed an identity rule that the earlier grouping had ignored.
“Latest” is a business rule with a sorting expression
In this lab, sequence is a non-negative BIGINT supplied by the producer. It orders versions within each normalized key. Version 2 takes precedence over version 1, regardless of arrival order.
Ingestion time is not automatically a safe substitute for a source sequence. A delayed version can arrive after a newer one. Sorting by arrival would then give the old state an undeserved promotion.
If your source uses a timestamp, define what happens when timestamps tie. A source sequence can resolve the tie if it represents a meaningful ordering. A random identifier may make the result repeatable, but repeatable does not necessarily mean correct.
For this example, the rule is deliberately strict: the same key and sequence must describe the same value. Two different values at the same position in the source history are a conflict. We reject the batch instead of inventing a winner.
Separate an exact retry from a contradiction
The implementation follows four steps:
- Validate the required types and reject missing or blank keys and invalid sequences.
- Normalize the key according to the lab’s identity rule.
- Collapse exact repeats of key, sequence, and value, then reject multiple values for one key and sequence.
- Select the highest sequence for each key.
The distinction matters. dropDuplicates(['key']) expresses which columns define duplication; it does not express a newest-version policy. Spark documents deduplication separately from the ordering you must define for your business rule. PySpark reference.
After conflict validation, the selection itself is small:
from pyspark.sql import Window, functions as F
order = Window.partitionBy("key").orderBy(F.col("sequence").desc())
prepared = (
unique.withColumn("_rank", F.row_number().over(order))
.where("_rank = 1")
.drop("_rank")
)
Here, unique means the already-normalized and validated batch. Without that validation, row_number would silently pick one of the conflicting rows. A window function is very good at applying a rule; it cannot supply the missing business decision.
Add a fourth row, A | 2 | other, and the batch is rejected: same key, same sequence, different value.
The complete implementation is available in the Data Engineering Lab, including the checks around this selection.
- Exact retry (same key, sequence, value): collapsed.
- Same key and sequence, different value: batch rejected.
- Higher sequence within the batch: wins.
- Older than target: ignored by the illustrated MERGE below.
Exact retries can collapse. Contradictory versions need a decision. Original AI-generated editorial illustration.
NULL deserves an explicit policy too
The lab permits a NULL payload and does not interpret it as a delete. That choice is intentional. Some systems use null to mean an absent attribute, others use it to mean a cleared value, and others encode deletions in a separate operation field.
A NULL value and a text value at the same key and sequence still conflict here. The implementation removes exact duplicate rows, then counts distinct rows per key and sequence. That avoids relying on an aggregate over distinct payload values that could ignore NULL.
An empty DataFrame returns an empty result with the expected schema. A streaming DataFrame is rejected: the function accepts a finite batch and performs Spark actions during validation.
That last detail has an operational consequence. Use a stable input snapshot. If the underlying source can change between validation and evaluation, the function can validate one view of the input and later evaluate another. The contract assumes the batch remains the same batch.
Run the example without a cloud account
Clone the repository and run these commands from its root with Python 3.12 and Java 17 available:
python -m pip install -r pyspark/day_001_cdc/requirements.txt
python -m unittest discover -s pyspark/day_001_cdc -p 'test_*.py' -v
PySpark is pinned to 3.5.3. On Windows, the README explains the local Java and networking settings used for validation.
Eight tests pass against a real local Spark JVM. They cover normalized keys, exact retries, conflicting versions, NULL handling, invalid inputs, empty input, repartitioning, and idempotence of the preparation.
A well-prepared batch can still overwrite the wrong state
Suppose the target already contains version 10, and a later run receives version 8. The batch preparation correctly selects version 8 from that batch. The write must still prevent the persisted state from moving backward.
An illustrative Delta adaptation is:
MERGE INTO sandbox.customer_current AS t
USING prepared_updates AS s ON t.key = s.key
WHEN MATCHED AND s.sequence > t.sequence THEN
UPDATE SET t.sequence = s.sequence, t.value = s.value
WHEN NOT MATCHED THEN
INSERT (key, sequence, value)
VALUES (s.key, s.sequence, s.value);
This assumes a target with unique normalized keys and non-null sequence values. Equal versions with conflicting source and target payloads require an additional validation policy. The lab detects conflicts inside a batch, not between that batch and an already-persisted target.
Runtime details matter as well. In DBR 15.4 LTS and below, duplicate-match detection considers the ON condition. In DBR 16.0 and above, it also considers the WHEN MATCHED conditions. Record your runtime and reproduce the actual case there. MERGE documentation.
Preserve history when history is the requirement
Keeping the last event per key is appropriate for some current-state workloads. It is not a general CDC algorithm. If a consumer needs every intermediate transition, discarding those transitions loses information even when the final state is correct.
Before promoting this pattern, test a repeated batch, an older incoming version, a conflicting version already in the target, and the intended delete behavior. Check the target’s uniqueness assumption and your concurrency model. Measure shuffle and validation costs using representative data rather than treating a small correctness test as a scale claim.
The useful question is not “How do I make MERGE stop complaining?” It is “Can I explain exactly which record should win, and why?”
Once that answer is explicit, the code has a job it can actually perform. Mind reading can remain outside the service-level agreement.
Scope: synthetic data, validated on local Spark 3.5.3. The MERGE example is illustrative and was not run against Delta. It does not cover deletes, concurrent writers or SCD Type 2 history.
References checked on October 18, 2026.
Top comments (2)
The point that a window function applies a rule but can't supply the business decision is the part most "just use row_number" answers skip. One check I'd add to the conflict step: compare the payload with a hash of the value columns rather than the raw columns, for example sha2(to_json(struct(...)), 256), so a new nullable column or a float that round-trips differently doesn't turn an exact retry into a false conflict (or hide a real one). It also makes the rejection report easier to read, since you can show the key, the sequence and the two differing hashes. Normalising the key once and reusing that same column in the grouping and in the ON clause is a great habit too; I've seen the "normalise only inside ON" version cause exactly the multiple-match error you describe.
Thanks, Karan, I appreciate the careful read!🙂
The hash idea is a good one, mostly for the rejection report. Showing the key, the sequence and two short hashes is much easier to scan than comparing raw payloads side by side. Hashing an explicit list of value columns also means a column added later doesn’t silently change what counts as an “exact retry”.
I tested it on local Spark and added the code to the lab repo: github.com/jenniferevangelista-de/...
Two things came out of the tests. First, a float that round-trips differently still produces a different hash (0.1 + 0.2 vs 0.3), so I round to an agreed scale before hashing, and a real change (0.30 vs 0.31) is still caught. Second, the order of columns inside the struct changes the hash, so I pin an explicit, fixed column list in code.
I also kept the raw values next to the hashes in the report, so whoever investigates can see what actually differs.
And yes, normalising the key once and reusing that same column everywhere is the habit that saved me the most debugging time. Thanks for confirming you’ve seen the same failure.