Forum Discussion
Merge statement creates duplicates - Low shuffle merge as cause?
Hi TomMa ,
In this context, "cache the bronze DataFrame" means calling .cache() (or .persist()) on your bronze_filtered DataFrame before passing it into the merge. Something like:
def merge_data(table_name, bronze_filtered, target_table, id_col='Id'): bronze_filtered = bronze_filtered.cache() # Add this bronze_count = bronze_filtered.count()
he reason this matters: Spark DataFrames are lazily evaluated, meaning the computation that produces bronze_filtered is re-executed each time Spark needs the data. If your upstream pipeline has any non-determinism (e.g., reading from a source that changed, a non-deterministic UDF, unstable sort), the DataFrame could theoretically produce different rows when Spark materializes it during the merge vs. when you called .count() earlier.
By calling .cache(), you force Spark to materialize the DataFrame once and hold it in memory/disk, so the exact same rows are used for the count check and for the merge source. This rules out non-deterministic re-evaluation as a cause of the duplicates.This is a diagnostic step more than a fix — true upstream non-determinism would be the real bug to fix.
If this post helps, then please appreciate giving a Kudos or accepting as a Solution to help the other members find it more quickly.
If I misunderstand your needs or you still have problems on it, please feel free to let us know. Thanks a lot!
Super UserHi TomMa ,
The metric combination you described — numSourceRows: 1 producing numTargetRowsInserted: 5 with numTargetRowsCopied: 0 — genuinely shouldn't be possible under standard merge semantics. A single unmatched source row can only ever produce one insert. The only way you get 5 inserts from 1 source row is if the merge engine is misidentifying already-existing target rows as "not matched" and re-inserting them. That's exactly the class of bug Low Shuffle Merge could introduce, because its core optimization involves partitioning which rows need full rewrite processing vs. which don't — and a misclassification there could cause rows to fall into the wrong bucket.
Before touching Spark configs, I'd verify these since they're lower risk to investigate:
Duplicate IDs emerging during the bronze filter
Schema evolution or partition mismatches
The ID column having leading/trailing whitespace or case sensitivity issues.
The most productive path is probably: cache the bronze DataFrame, rerun for a week, and see if duplicates persist before touching the Spark config. If they do, then disabling Low Shuffle is a reasonable next step to isolate the cause.
If this post helps, then please appreciate giving a Kudos or accepting as a Solution to help the other members find it more quickly.
If I misunderstand your needs or you still have problems on it, please feel free to let us know. Thanks a lot!
