Spark Join Strategies

Topics Covered

Four Ways to Join, and What Chooses Between Them

The same word, four implementations

Broadcast hash join

Shuffle hash join

Sort merge join

Broadcast nested loop join

What the decision actually depends on

What this lesson covers

Reading which one you got

A join is one word in your query and four entirely different algorithms underneath it. The engine picks one of them from numbers it estimated before reading anything, and that single choice is usually the difference between a query that finishes in a minute and one that does not finish at all.

The same word, four implementations

Every distributed join has to solve the same problem: rows that must be compared are sitting on different machines. There are only two ways to fix that. Either you move one side everywhere so that every machine has all of it, or you move both sides so that matching keys land on the same machine. Everything else is detail.

Moving one side everywhere is a broadcast. It costs one copy of that side per executor and no shuffle at all. It is only possible when that side is small enough to fit in an executor's memory several times over, because every concurrent task on that executor may hold a reference to it.

Moving both sides is a shuffle. Rows are partitioned by a hash of the join key and written to disk, then read back by the task responsible for that key range. It works at any size and it costs a full redistribution of both inputs across the network.

Within each of those two approaches there is a further choice about how the matching itself is done. A hash join builds a lookup table from one side and probes it with the other. A sort merge join sorts both sides and walks them in step. The first is faster when the build side fits in memory and the second is the only option when it does not.

That gives four strategies, and knowing which one you got is the single most useful thing a query plan tells you.

The two ways to get matching rows onto the same machine. One copies a small side everywhere, the other redistributes both sides. Everything else about joins is a detail inside these two.

Broadcast hash join

One side is collected to the driver, serialised, and sent to every executor. Each executor builds a hash table from it once and keeps it for the duration of the stage. The other side is then scanned where it already lives, and every row probes the hash table.

Nothing is shuffled. The large side is never moved, never sorted, and never written to disk. For a fact table joined to a small dimension this is not merely faster, it is a different order of cost, because the dominant expense of a join at scale is the redistribution it avoids entirely.

The constraint is memory. The broadcast copy exists once per executor, and every task on that executor shares it, which is good. But it is built first on the driver, which must hold the whole thing while it collects and serialises it, and that is where broadcast joins actually fail. A driver with four gigabytes of heap cannot broadcast a two gigabyte table, whatever the executors could manage.

Shuffle hash join

Both sides are shuffled by the join key, so matching keys land in the same partition. Then, within each partition, the smaller side is used to build a hash table and the larger probes it.

This is the right strategy when one side is too large to broadcast but each individual partition of it is small enough to build a hash table from. It pays the shuffle but avoids the sort, and avoiding the sort is worth a great deal because sorting is where joins spill.

It is also the strategy engines are most reluctant to choose, because the consequence of getting it wrong is severe. If a partition's build side does not fit in memory, there is no graceful fallback inside the operator. Most engines therefore require a clear size advantage before selecting it, and many default to disabling it entirely.

Sort merge join

Both sides are shuffled by the join key and both sides are sorted within each partition. The operator then walks the two sorted streams in step, advancing whichever is behind, emitting matches as it goes.

It needs almost no memory, because it holds only the current position in each stream plus whatever rows share the current key. That is why it is the default. It cannot run out of memory in the way a hash build can, and when it does need more room than it has, the sort beneath it spills to disk and carries on correctly.

The cost is the sort. Sorting is superlinear, it writes to disk when the data does not fit, and it has to complete on both sides before the first output row can be produced. On two large inputs it is the right trade. On a large input joined to a small one it is enormous waste.

Broadcast nested loop join

The fourth strategy is the one nobody chooses deliberately. When the join condition is not an equality, there is no key to hash or sort on, so the engine broadcasts one side and compares every row against every row.

It is quadratic. A hundred thousand rows against a hundred thousand rows is ten billion comparisons. The plan node is easy to spot once you know to look for it, and seeing it almost always means a condition got written as an inequality or a function when an equality was intended.

What the decision actually depends on

The engine chooses by estimated size, in a fixed order. If either side's estimate is below the broadcast threshold, broadcast. Otherwise, if the join is an equality and one side is much smaller than the other and shuffle hash joins are enabled, shuffle hash. Otherwise, if the keys are sortable, sort merge. Otherwise, nested loop.

Everything in that sequence hangs on the word estimated. The planner uses statistics if they exist, file sizes if they do not, and a configured default if neither is available. A table with no collected statistics stored as compressed columnar files is routinely estimated at several times its real size, because the planner sees the uncompressed logical width rather than the bytes.

This is why collecting statistics on the dimension tables is the single highest value thing you can do for join performance. It is not a tuning parameter. It is giving the decision the input it needs.

What this lesson covers

The sections that follow take the three strategies that matter in order. Broadcast joins and the threshold that governs them, including why the threshold is a soft limit rather than a hard one. Sort merge join and the two shuffles and two sorts it implies. Skew, which is the failure mode that no strategy choice can fix, and the remedies that actually work. And finally how to force a strategy, how to verify what you got, and when forcing is the wrong answer.

Reading which one you got

All four strategies appear in the physical plan under names that describe them literally, and the name is the first thing to find in any join investigation.

A broadcast hash join shows a broadcast exchange beneath one side and nothing beneath the other, which is the visual signature worth memorising: one side has an exchange, the other does not. A sort merge join shows a hash exchange and a sort beneath both sides, four nodes where the broadcast version had one. A shuffle hash join shows the two exchanges without the sorts. A nested loop join names itself and has no key in its condition.

Two numbers beside those nodes complete the picture. The estimated size on each side is what the decision was made from, and comparing it against the bytes the stage actually read tells you immediately whether the planner was working from anything real. A side estimated at eight gigabytes that read four hundred megabytes is a broadcast that should have happened and did not. A third number completes it. The output row count against the two input row counts tells you whether the join exploded. A join that emits far more rows than either side brought in has duplicate keys on both sides, and that is a correctness question wearing a performance costume. It is also the most common cause of a join stage that runs for hours with no skew in its inputs, because the explosion happens inside the operator where no input metric shows it.

Key Insight