In a Spark join, how does salting a hot key spread its rows across many tasks?

## The problem A shuffle join hash-partitions both sides on the join key. If one key value carries a large share of the rows, all of those rows land in one partition, one task reads them all, and the stage waits. No amount of extra parallelism helps, because the key is atomic: `hash("c-42")` has exactly one answer. Salting attacks the cause. If the key is what concentrates the data, make the key artificially wider. ## The recipe 1. Choose a salt count `N`. 2. On the **large** (skewed) side, add a salt column with a random integer in `[0, N)`. 3. On the **small** side, replicate each row `N` times, once per salt value — typically by adding a literal array of `0..N-1` and exploding it. 4. Join on `(original_key, salt)` instead of `original_key`. 5. Drop the salt column from the result. After step 4 the hot key has become `N` distinct composite keys, which hash to up to `N` different partitions, so up to `N` tasks share the work. ## Why the result is still correct This is the part interviewers probe. Each large-side row is assigned exactly one salt, so it exists once. Each small-side row exists in all `N` salted copies, so whichever salt the large-side row drew, there is a matching copy carrying it. Every pair that would have matched in the unsalted join matches exactly once in the salted one — no duplicates, no drops. The classic mistake is salting **both** sides randomly. Then a large-side row with salt 7 and its matching small-side row with salt 3 never meet, and rows silently disappear from the output. "Salt one side randomly, replicate the other exhaustively" is the invariant. ## Choosing N Size it from the data, not from taste: take the hot key's total bytes and divide by the partition size you want each task to handle, rounding up. If one key holds 20 GB and you want ~200 MB tasks, `N` is about 100. Going larger costs you: the replicated side's row count is multiplied by `N`, and shuffled accordingly, so an `N` of 1000 against a 10-million-row dimension means ten billion rows in the exchange. In practice `N` between 8 and 64 covers most cases. ## Salting only the hot keys Salting every key multiplies the small side by `N` for nothing — the well-behaved 99.9% of keys were never a problem. The refined version salts only the values you have measured as hot: the large side gets a random salt for hot keys and a constant `0` for everything else, and the small side is split into a "cold" copy with salt `0` and a "hot" copy exploded across all `N` salts, then unioned. Note that a generator such as `explode` cannot be nested inside a `when` expression in Spark, which is why the small side is built by filtering and unioning rather than by a single conditional column. The drawback is that the hot-key list is a hard-coded assumption about the data. It rots. Either compute it at runtime from a cheap count of the key, or schedule a review of it — a salted job whose hot-key list is a year out of date is skewed again, and nobody notices until it misses an SLA. ## Salting an aggregation For a skewed `groupBy` — which the adaptive skew-join rule does **not** cover — the equivalent is two-phase aggregation: group by `(key, salt)` to compute partials spread across `N` tasks, then group the partials by `key` alone to combine them. This works cleanly for associative, commutative aggregates like `sum`, `count`, `min` and `max`. It does not work for aggregates whose partials cannot be merged, such as an exact median or an exact distinct count, where you need a different approach (approximate sketches, or a dedicated pass over the hot group). ## When *not* to salt Salting is real code in your pipeline, with real correctness risk, so exhaust the cheaper options first: - **Can you broadcast the small side?** That removes the shuffle entirely and no salt is needed. - **Is the hot key a null or a sentinel?** Nulls never match in an inner equi-join; filtering or handling them separately usually costs one predicate. - **Is AQE's skew-join splitting already handling it?** For a shuffle join whose skewed partition clears the factor and the byte threshold, Spark splits it for you at runtime. What salting still buys you when those fail: it changes the distribution **before** the shuffle is written, so the map side, the shuffle write and the fetch are all balanced — whereas the adaptive rule only redistributes the read of an already-written giant partition. And it covers aggregations, which the adaptive rule does not touch at all.