How a Join Is Planned

orders.join(customers, on="customer_id") was one line in the Rows That Match lesson. On a cluster it is one of four different jobs, and Spark chooses which from two facts: how big each side is, and whether the join condition is an equality. The plan tells you which one you got, and one of the four is the thing to fear.

Watch it happen

The Daily Grind's customers, their orders, and a table of price bands:

The join you know, with the plan trimmed to its join line:

joined = orders.join(customers, on="customer_id")
joined.explain()
result = joined

HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(#1@1, #3@0)]

Sail collects one side whole and looks the other up in it, row by row: no shuffle, the shape a cluster calls a broadcast hash join. Asking for it changes nothing here, because it is already what you got:

joined = orders.join(F.broadcast(customers), on="customer_id")
joined.explain()
result = joined

Now a join with no equality in it. Which band is each order in?

banded = orders.join(bands, (orders.price >= bands.low) & (orders.price < bands.high))
banded.explain()
result = banded.select("order_id", "price", "band")

NestedLoopJoinExec: join_type=Inner, filter=#2@0 >= #4@1 AND #2@0 < #5@2

A different operator. There is no key to hash, so every order is compared with every band: four orders, three bands, twelve comparisons. Four million orders and three thousand bands is twelve billion.

The four

Broadcast hash join. The small side is sent whole to every executor, and the big side is looked up against it, partition by partition. No shuffle. Spark chooses it on its own when a side is under spark.sql.autoBroadcastJoinThreshold, 10 MB by default; F.broadcast(df) asks for it, and the docs promise it is honoured "even if the size of table … suggested by the statistics is above the configuration."

Sort-merge join. Both sides shuffled on the key, sorted, and walked together. Two shuffles and two sorts; scales to any size; the default for two big tables with an equality condition.

Shuffle hash join. Both sides shuffled on the key, and on each executor the smaller partition becomes a hash table. No sort, more memory.

Broadcast nested loop join. When the condition is not an equality, between, >, a like, there is no key to hash or sort on, so every row of one side is compared with every row of the other. It is the one that explodes.

Hints, and who wins

Spark's hint names are BROADCAST, MERGE, SHUFFLE_HASH and SHUFFLE_REPLICATE_NL, and when both sides carry different hints, "Spark prioritizes the BROADCAST hint over the MERGE hint over the SHUFFLE_HASH hint over the SHUFFLE_REPLICATE_NL hint." Since Spark 3.2, adaptive query execution can also change its mind after the shuffle has shown it the real sizes: it "converts sort-merge join to broadcast hash join when the runtime statistics of any join side are smaller than the adaptive broadcast hash join threshold."

What this site shows

Every equi-join here plans as HashJoinExec: mode=CollectLeft, the broadcast shape, with or without the hint, and every join without an equality plans as NestedLoopJoinExec. You can see the fork. You cannot see a sort-merge join, because one process never needs one.

The symptom, and the fix

A join between two modest tables takes longer than everything else combined. Read the plan. BroadcastNestedLoopJoin means the condition has no equality in it and the join is all pairs: rewrite it so that an equality is the join condition and the range is a filter after it, or bucket the range into a key you can equi-join on. SortMergeJoin with one side small means Spark's estimate of that side is missing or wrong: broadcast it. A broadcast that fails with an out-of-memory on the driver means the "small" side was not small; let it be a sort-merge.

Self-check

orders.join(rates, orders.day.between(rates.start, rates.end)) is slow. Why, and what first?

No equality in the condition, so it is a nested loop comparing every order with every rate row. Join on something equal first, a currency or a year, and keep between as the filter after.

Nothing to grade here. Four lessons, one habit: a plan, its pieces, a hot key, and a join's shape are all in explain() before you touch a cluster, so read it first. Next, Keep It Around: what happens when five reports need the same expensive DataFrame.