Apache Spark WTF??? 😎 Straight Outta Comparison ⚖️

“You are now about to witness the strength of Spark street knowledge”
(adapted from “Straight Outta Compton” by N.W.A.)

---

🧪 Lab Time: Let’s Manufacture a 100-Million-Row Crime Scene

Instead of stuffing a giant notebook into the article like a 300-line lasagna, the full script lives in GitHub: 0__generate_test_data.py.

The goal is simple: create two large Spark DataFrames, make them look mostly alike, secretly damage one of them, and then use them later to test full-row comparison strategies.

The Lab Factory: Homemade Spark SQL Mischief

The dataset is generated with plain Spark: spark.range(), Spark SQL functions, deterministic hashes, seeded random values, and a couple of helper functions.

We generate df_a: a 100-million-row DataFrame with 100 partitions and 10 business-style columns:

root  
 |-- uuid: string (nullable = false)  
 |-- pk_part_1: string (nullable = false)  
 |-- pk_part_2: string (nullable = false)  
 |-- pk_part_3: string (nullable = true)  
 |-- amount_double: double (nullable = true)  
 |-- score_float: float (nullable = true)  
 |-- quantity_int: integer (nullable = true)  
 |-- status_int: integer (nullable = false)  
 |-- created_ts: timestamp (nullable = true)  
 |-- updated_ts: timestamp (nullable = true)

The dataset mixes strings, pseudo-key parts, doubles, floats, integers, statuses, and timestamps. The pseudo-key columns are generated using hash-based deterministic choices, while numeric and timestamp fields use seeded randomness. So it is synthetic, but not completely boring.

After generation, df_a is written to Parquet and read back immediately. That is intentional. It materializes the data on disk without trying to cache 100 million rows in memory like a laptop pretending to be a warehouse cluster.

Why This Dataset Looks the Way It Looks

The goal is not random chaos. The goal is known, intentional damage so we can verify whether the comparison catches it.

df_b is built from df_a, but a hash bucket decides each row’s fate:

buckets < 10:    removed rows  
buckets 10..19:  modified rows  
buckets >= 20:   unchanged rows

The modified rows get changes in fields like amount_double and status_int. For a full-row comparison, this matters: Spark does not know that the same business entity changed. It only sees one old row disappear and one new row enter wearing a fake moustache.

Then the script adds around 100,000 brand-new rows that exist only in df_b, with new pseudo-key values, different status values, larger amount ranges, and 2027 timestamps. Finally, df_b is written to Parquet too, giving both suspects the same kind of physical evidence trail.

Why This Is Still Not Production-Messy Enough

This lab gives us scale, known differences, Parquet-backed evidence, two-way row-level diffs, and duplicate-aware comparison material. Big enough to make Spark sweat, controlled enough to know what should happen, and durable enough to survive beyond a nervous notebook cell.

What it does not give us is the full production circus: business-key change attribution, dirty source behavior, real skew, schema drift, JDBC single-threaded reads, or nested-data horror. Those goblins arrive later.

---

🔨 Equality Functions: The Tiny Hammer With a Fake Badge

Spark has a few functions that sound like they compare DataFrames. They do — but only in very specific ways. Some compare logical plans. Some compare test outputs. None of them replace a production reconciliation framework.

What This Test Is About

The script 1__equality_functions.py asks one question: what do Spark’s built-in equality helpers actually compare?

Not all “equality” means the same thing. assertSchemaEqual checks schema shape, sameSemantics checks logical-plan similarity, semanticHash fingerprints the plan, and assertDataFrameEqual compares small expected-vs-actual DataFrames in tests. None of them are production diffs.

The script loads df_a and df_b from Parquet, checks that their schemas match, then shows that same schema does not mean same plan or same data. df_a.sameSemantics(df_a.select(df_a.columns)) is the “same recipe” case, while df_a.sameSemantics(df_b) will usually be false because both DataFrames come from different Parquet paths.

Finally, it tests assertDataFrameEqual on small samples: equal data passes, modified data fails, reordered rows depend on checkRowOrder, tiny numeric drift depends on tolerance, and reversed columns pass only when column order is explicitly ignored.

How They Work

Spark’s equality-related functions mostly split into two groups:

1. Plan Equality Functions

• sameSemantics() and semanticHash() were introduced in Spark 3.1.0 and gained Spark Connect support in Spark 3.5.0.
• sameSemantics() checks whether two DataFrames have semantically equivalent logical plans, tolerating cosmetic differences such as attribute names. semanticHash() returns a compact hash of that simplified logical plan.

Both compare the plan, not the materialized data. They do not prove rows match, detect duplicate drift, find missing keys, or produce metrics and traces. Useful for developer tooling and query-generation checks; not for business reconciliation.

2. Testing Equality Functions

• assertDataFrameEqual() compares schema and row content with options for row order, numeric tolerances, nullable checks, column order/name handling, and diff diagnostics. It is great for unit tests, notebooks, CI, and controlled transformation checks — but it is still an assertion tool, not a reconciliation report.
• assertSchemaEqual() compares schemas for contract and regression tests. Useful, but schema equality is not data equality: the columns can be perfect while the values are on fire.

What Changed in Spark 4.x?

Spark 4.x improves equality testing, especially for pandas and pandas-on-Spark. Spark 4.0 introduced pyspark.pandas.testing.assert_frame_equal and expanded assertDataFrameEqual() support to pandas DataFrames.

It also introduced SQL collation support, reminding us that string equality can depend on case, accents, locale, and ordering rules. The toolbox got better, but Spark still does not provide a production-grade DataFrame diff API.

What They Are Good For

• Use sameSemantics() for logical-plan equivalence and query-generation checks.
• Use semanticHash() for plan fingerprinting and tracking query shapes.
• Use assertDataFrameEqual() for unit tests, notebook checks, CI, and small controlled outputs.
• Use assertSchemaEqual() for schema contract and regression tests.

Where They Fall Short

They are useful when the question is small, controlled, and test-shaped: unit tests, CI checks, notebook validation, and transformation regression tests. In short, they are great for controlled validation, not for production-scale reconciliation.

👮 Final verdict: For large production comparisons, you still need bigger strategies: exceptAll, union plus group counts, full outer joins by key, changed-cell extraction, grouped comparisons, hash-assisted approaches, and forensic metrics.

---

🚪 exceptAll: The Duplicate-Aware Bouncer

What This Test Is About

The script 2__exceptAll.py starts with a full-row question: do df_a and df_b contain the same complete rows, with the same duplicate counts?

That is why we run the comparison in both directions:

df_a_minus_df_b = df_a.exceptAll(df_b)  
df_b_minus_df_a = df_b.exceptAll(df_a)

df_a_minus_df_b contains rows that exist more often in df_a: removed rows plus the original versions of modified rows. df_b_minus_df_a contains rows that exist more often in df_b: brand-new rows plus the modified versions.

If amount_double or status_int changes, exceptAll does not say, “same pseudo-key, changed payload.” It says, *“old full row disappeared, new full row appeared.”*Brutal, correct, not very chatty.

Both diff outputs are written to Parquet and read back before counting, because durable evidence beats notebook theatrics.

How exceptAll Works

exceptAll asks: which full rows appear in the left DataFrame more times than in the right?

It is duplicate-aware: if a row appears three times in df_a and once in df_b, the result contains that row twice.

Conceptually, it computes:

result_count = count_in_df_a - count_in_df_b

and emits only rows with a positive count.

Under the hood, Spark usually turns this into a counting problem: tag df_a rows with +1, tag df_b rows with -1, union both sides, group by the full row, sum the tags, keep positive balances, and emit each row that many times.

Elegant, yes. Free, no. Grouping full rows usually means shuffle. Wide rows hurt. Skew hurts more. Running it both ways means two directional investigations.

Also, exceptAll compares columns by position, not by name. In this lab, that is safe because df_b is generated from df_a, so column order is preserved. In real production data, normalize column order first. Explicit column selection is cheap. Debugging a positional mismatch in a 100+ column table is not.

What exceptAll Is Good For

Use exceptAll when the contract is: The whole row is the unit of truth, and duplicates matter.

It is a strong baseline for snapshot comparison, pipeline output validation, generated table comparison, backup/export validation, and duplicate-aware reconciliation.

Two outputs tell the story:

df_a_minus_df_b = what A has extra  
df_b_minus_df_a = what B has extra

If both are empty, the DataFrames are equal as multisets of full rows. That is a strong full-row equality statement.

It is also safer than subtract when duplicates matter, because subtract behaves like distinct set difference. exceptAll keeps the duplicate goblin in custody.

Where exceptAll Falls Short

exceptAll is blunt. It does not tell you which column changed, which business key was affected, whether a float is within tolerance, whether timestamps need normalization, or why nulls are being weird.

If one row changes from:

amount_double = 100.00  
status_int    = 1

to:

amount_double = 1099.99  
status_int    = 9

exceptAll only reports:

the original full row exists only in df_a  
the modified full row exists only in df_b

Correct, but not enough for a human incident report. For “same pseudo-key, these columns changed,” you need a key-based diff.

It also returns rows, not a compact metric report. If you want df_a_count, df_b_count, delta, and duplicate summaries, the union + group count strategy is often better.

👮 Final verdict: exceptAll finds full-row imbalance with duplicate awareness. It finds the bodies. It does not write the incident report.

---

🔑 Full Outer Join: The Key-Based Interrogation Room

What This Test Is About

The test 3__full_outer_join.py asks a different question from exceptAll.

exceptAll asks: *Do both DataFrames contain the same complete rows?.*The full outer join asks: For the same business key, did the payload change?

In this lab, the key is uuid. Since df_b is built from df_a, unchanged and modified rows keep the same uuid, while removed rows disappear from df_b and brand-new rows appear only in df_b.

That lets us classify every key into one of four buckets: == (equal), << (left only key), >> (right only key) and <> (changed).

So instead of saying “old full row disappeared, new full row appeared,” this strategy can say for example: same uuid, amount_double changed, status_int changed.

That is the upgrade: from row-level mismatch to key-level explanation.

How It Works

First, the script loads df_a and df_b from Parquet.

Then it defines the comparison contract:

key columns:      uuid  
payload columns:  every non-key column

The key column stays unmodified. Every payload column gets a suffix: _a for the left side and _b for the right side. This lets both versions of the same column survive after the join.

Then the script performs a full outer join on uuid.

A full outer join keeps everything:

keys only in df_a  
keys only in df_b  
keys present in both

To detect whether a key existed on each side, the script adds side markers before joining: _present_a and _present_b.

After the join, each payload column is compared with null-safe equality. That means:

null vs null      = equal  
null vs value     = different  
value vs value    = normal comparison

Then the script assigns a row-level status:

<<: key exists only in df_a  
>>: key exists only in df_b  
<>: key exists in both, but at least one payload column differs  
==: key exists in both, and all payload columns match

The script writes three outputs:

• full outer check: the “source of truth” join containing every key, every flag, and both versions of the payload.
• column stats: a match percentage report per column (safeguarded against division-by-zero if your input tables are empty).
• KOs: the “Wall of Shame.” It shows only rows that are not equal, dynamically filtered to only show columns that actually contain a mismatch.

That last output is the important one. It turns a wide diff into a human-readable trace.

Instead of forcing someone to inspect 100 columns side by side, it says:

uuid = ...  
column_name = amount_double  
value_a = ...  
value_b = ...

Much easier to explain. Much harder to hide from.

📝 Note: To keep the investigation focused, the KOs output doesn’t just filter the rows; it dynamically trims the columns. If a column is 100% equal across the whole dataset, it is dropped from the KO view entirely. This keeps your “wide” diff from becoming a “300-column-wide” nightmare.

What It Is Good For

The full outer join strategy is best when your data has a reliable business key and you need to explain differences by entity.

It is good for questions like:

Which keys are missing from df_b?  
Which keys are new in df_b?  
Which keys exist in both but changed?  
Which columns changed?  
What were the values before and after?

This makes it much more useful than exceptAll for business-facing reconciliation.

exceptAll is great when the whole row is the unit of truth. But if the business thinks in terms of entities — customers, orders, transactions, accounts, events — then a key-based diff is usually more useful.

The changed-cells output is especially valuable because it gives a compact trace: key, column_name, value_a, value_b.

That is exactly the kind of output people need for debugging, incident reports, migration validation, and “please explain why the dashboard changed” meetings.

It also separates different kinds of problems:

missing on the left  
missing on the right  
same key, changed payload  
fully equal

That separation is gold. It tells you whether you have a missing-data problem, an extra-data problem, or a value-drift problem.

Where It Falls Short

A full outer join is only as good as its key. If uuid is unique and stable, great. If the key is duplicated, unstable, nullable, or not actually a business identifier, the join can become misleading or explosive.

Duplicate keys are the big danger. If one key appears twice in df_a and three times in df_b, a normal join creates a many-to-many match. That can multiply rows and produce nonsense-looking diffs.

So before trusting this strategy, check key health:

Are keys unique in df_a?  
Are keys unique in df_b?  
Are any keys null?  
Are the keys actually business-stable?

Another limitation: this strategy compares payload columns directly. If you need tolerance for floats, timestamp normalization, string trimming, case-insensitive comparison, array sorting, or custom rules, you must add those rules explicitly.

It can also be expensive. A full outer join on large DataFrames usually means shuffle. Wide payloads make the output heavy. Changed-cell extraction can also create a lot of rows if many columns differ.

And finally, this strategy is not a duplicate-aware full-row comparison. It does not replace exceptAll. It answers a different question:

exceptAll:        do the full rows balance?  
full outer join:  what happened per key?

Different weapon. Different interrogation room.

👮 Final verdict: Use full outer join when the question is key-based and you need missing, new, and changed entities with column-level evidence. But validate the key first, handle duplicates, normalize values, and expect a shuffle bill. exceptAll finds row imbalance; full outer join explains entity drift.

---

🙋🏽‍♂️ Native Cell Diff With Thresholds: The Comparison Report Humans Actually Wanted

What This Test Is About

The script 4__full_outer_plus_union.py asks: For the same candidate key, which values are different?

The candidate key is: pk_part_1, pk_part_2 and pk_part_3.And the output is a narrow diff table:

pk_part_1  
pk_part_2  
pk_part_3  
column_name  
_diff_status  
value_a  
value_b

Only differences are emitted. Equal rows stay quiet, which is exactly what we want. Nobody needs a 100-million-row report saying “still fine, still fine, still fine.”

The compact statuses are:

<<  exists only in df_a  
>>  exists only in df_b  
<>  exists in both, but differs

We also add explicit thresholds for noisy fields. amount_double and score_float get numeric tolerances, while created_ts and updated_ts get timestamp tolerances. So the script is not doing blind exact equality everywhere; it compares each column according to the contract.

How It Works

The script starts by selecting the candidate key columns and the columns to compare. It also defines tolerance rules:

amount_double: ±0.01  
score_float:   ±0.0001  
created_ts:    ±1 second  
updated_ts:    ±1 second

Numeric columns are compared with absolute tolerance. Timestamp columns are compared with second-level tolerance using unix_timestamp(). Everything else uses null-safe equality, so null versus null is treated as equal, while null versus a value is not.

Before comparing values, the script performs a key health check. It counts how many rows each candidate key has in df_a and df_b.

Keys are split into two lanes:

safe keys:  
  at most one row in df_a  
  at most one row in df_b  
non-unique keys:  
  more than one row on either side

For safe keys, the script performs a full outer join. Because there is at most one row per key on each side, the join cannot create many-to-many nonsense. Then it compares every payload column using the right rule: tolerance-based comparison for configured numeric and timestamp fields, and null-safe equality for the rest. Finally, it explodes only the failing comparisons into rows.

So if one key has two changed columns, the output has two rows:

same key | amount_double | <> | value_a | value_b  
same key | status_int    | <> | value_a | value_b

But if amount_double differs only inside the allowed threshold, it does not appear in the diff. Same for timestamp drift inside the configured tolerance.

For missing safe-key rows, there is no specific changed column, so the script emits one row with:

column_name = "__ROW__"

and stores the full existing payload as JSON in value_a or value_b.

For non-unique candidate keys, the script refuses to invent a fake one-to-one match. Instead, it switches to a union + group strategy. It groups by:

candidate key + full payload

and compares how many times each payload appears in df_a and df_b.

Any imbalance becomes:

column_name = "__PAYLOAD__"

This means: “the duplicated-key group does not balance, but we are not pretending to know which row pairs with which.”

What It Is Good For

This strategy is good when you want detailed, explainable differences while staying inside Spark’s DataFrame engine.

It avoids Python UDFs, RDD cogroups, pandas applyInPandas, and Arrow worker problems. Everything is built from Spark-native joins, unions, groups, projections, structs, JSON, and column expressions.

The output is also genuinely useful. It gives you:

which key failed  
which column failed  
whether it was left-only, right-only, or changed  
what value was seen in df_a  
what value was seen in df_b

The threshold rules make it more realistic than strict equality. Floating-point noise and tiny timestamp drift do not automatically become incidents. The comparison contract decides what counts as real difference.

It also handles candidate-key ambiguity honestly. If the key is one-to-one, you get exact cell-level diffs. If the key is duplicated, you get duplicate-aware payload-count diffs instead of fake changed-cell stories.

This makes it strong for migration validation, source-vs-target reconciliation, regression checks, pipeline rewrites, and any situation where someone will eventually ask, “Okay, but what exactly changed?”

Where It Falls Short

This is still not free. The script uses a full outer join for safe keys and union + group for duplicated keys, so Spark still has to shuffle data.

The candidate key also controls how detailed the result can be. If the key is unique, the script can show changed columns directly. If the key is not unique, it falls back to __PAYLOAD__ rows because there is no safe row pairing without more business logic.

Tolerance rules only apply to the safe one-to-one path. For non-unique keys, payload comparison is exact, because applying tolerances inside duplicated groups would require a matching strategy first.

The __ROW__ and __PAYLOAD__ JSON values are useful evidence, but they can be wide. They are great for traceability, less great for pretty screenshots.

One more important detail: the stats are diff-row stats, not key stats. If one key has two changed columns, it contributes two <> rows. That is not overcounting; that is the whole point of a cell-level diff.

👮 Final verdict: Use this strategy when you want production-shaped, human-readable diff evidence without leaving Spark-native DataFrame land. It gives exact changed-cell rows for safe one-to-one keys, honest payload-count diffs for duplicated candidate keys, and explicit numeric/timestamp thresholds so tiny acceptable drift does not pollute the report. It still pays the shuffle bill, but at least the receipt tells you exactly what changed.

---

👍 A Good Diff Starts With a Good Question

Comparison is not a Spark function. It is a contract. Before choosing count, exceptAll, joins, hashes, thresholds, or cell-level diffs, you need to answer the real question: what are we trying to prove? Same row count? Same schema? Same full rows? Same business keys? Same values within tolerance?

The best strategy is the one that matches the risk. In a migration project, you might start with schema checks and row counts, then use duplicate-aware full-row comparison for snapshots, and finally key-based cell diffs for critical tables like invoices, transactions, balances, or orders. Maybe amount_double can tolerate 0.01. Maybe timestamps can drift by one second. But IDs, statuses, and money-moving flags may need exact truth.

Don’t compare everything in the most expensive way just because Spark can. A forensic diff over millions of rows costs shuffle, runtime, storage, and attention. Sometimes a count is enough. Sometimes exceptAll is enough. Sometimes you need the full key, column, value_a, value_b evidence table.

So define equality before computing it. Spark will execute almost any comparison at scale, including the wrong one. Your job is to decide what “same” means, what differences matter, what tolerances are allowed, and what evidence humans need when the comparison fails.

---

👋 Greet & Git

It was genuinely interesting investigating how to improve this comparison challenge. Thanks to Alberto Manzano and Pablo Salafor throwing it at me. A challenge I thought I had already solved before, but like all good movie villains, it came back for more…

As always, the full repo is here. Clone it, break it, benchmark it, and enjoy yourself.

---

“Yo, Spark.
Whassup?
Tell ’em where you from!
Straight outta Comparison”
(adapted from “Straight Outta Compton” by N.W.A.)