Spark joins

01 · What it is

One join, three ways to run it

A join combines two DataFrames on a key. Spark can run that one logical join three physically different ways, and the choice decides whether the job moves gigabytes across the network or almost nothing. In one line each — the mechanisms are the deep dive in §3:

• Broadcast Hash Join (BHJ) — copy the small side to every executor and join locally; the big side is never shuffled.
• Shuffle Sort-Merge Join (SMJ) — shuffle both sides by key, sort, merge. The default for big-vs-big.
• Shuffle Hash Join (SHJ) — shuffle both, build an in-memory hash table on one side. A rarely-chosen middle ground.

analogy A join needs matching keys on the same executor. Either move the data to the keys (shuffle both sides — everyone walks to the sorting desk) or move a copy of the small table to every executor (broadcast — hand each worker their own lookup card). Every strategy is just a different answer to "how do we get matching keys together?"

02 · Why it matters

The shuffle is the most expensive thing a join can do

• A shuffle is the dominant cost. Moving rows across the network so equal keys land together means network I/O + disk spill + serialization + a stage boundary where every executor waits for the slowest. Avoiding it is the single biggest lever.
• The classic enterprise join is big fact ⋈ small dimension (orders ⋈ countries, events ⋈ device-lookup). If it broadcasts, the fact table is read once and never shuffled — often a 10× win. If it falls back to sort-merge, you shuffle the entire fact table for nothing.
• It's a constant interview probe: "Your join is slow — what do you check?" The answer starts with "which strategy did the planner pick, and why?" — read from df.explain() and the Spark UI. The how and the how-to-read-it are the next sections.

03 · How it works — deep dive

The three strategies, and how Spark chooses

Each strategy is a different trade between shuffling data and copying a small table everywhere. Below is the one canonical explanation of all three — the diagrams, the comparison table, and the gotchas later point back here rather than re-explaining. Read each as: the mechanism, when Spark picks it, and the trade-off.

The three strategies at a glance — the prose below unpacks each row; this is the map to scan first:

3.1 · Broadcast Hash Join (BHJ) — the fast path

• Mechanism. Spark collects the small side to the driver, ships a read-only copy to every executor (a BroadcastExchange), and each task probes that in-memory hash table while streaming its slice of the big side. The big side is never shuffled. (Because the build happens on the driver, an over-sized broadcast is really a driver risk — Lesson 03.)
• When Spark picks it. Automatically when one side's estimated size ≤ spark.sql.autoBroadcastJoinThreshold (10 MB in OSS), or when you force it with broadcast() / a BROADCAST hint. AQE (Lesson 05) can also flip a sort-merge join to broadcast at runtime once it sees the real post-filter size.
• Trade-off. Blazing fast and shuffle-free, but the small side must fit in the driver's and each executor's memory. Broadcast something too big → driver OOM / maxResultSize exceeded.

use case orders (2 TB) ⋈ country_codes (8 KB) — broadcast the country table; the 2 TB is read once, never shuffled.

3.2 · Shuffle Sort-Merge Join (SMJ) — the default for big-vs-big

• Mechanism. Both sides are shuffled by key (two Exchange nodes), each partition is sorted by key, then the two sorted streams are merged in lockstep — walk both, emit matches. Sorting is what lets the merge stream without holding either side fully in memory, so it scales to two enormous tables.
• When Spark picks it. The default for large-vs-large equi-joins, because spark.sql.join.preferSortMergeJoin = true. Chosen whenever neither side is broadcastable.
• Trade-off. Robust and memory-friendly (it streams, sorted), but you pay the full double shuffle + sort — the price of joining two big tables.

use case transactions (1 TB) ⋈ accounts (300 GB) on account_id — neither fits in memory, so sort-merge is correct.

3.3 · Shuffle Hash Join (SHJ) — the rare middle ground

• Mechanism. Both sides shuffled by key, then one side is built into a per-partition hash table (no sort) and the other probes it.
• When Spark picks it. Only when neither side is broadcastable and preferSortMergeJoin = false (or a SHUFFLE_HASH hint), and one side fits in memory per partition.
• Trade-off. Skips the sort (can beat SMJ when one side is moderately small), but the hash table must fit in execution memory per partition — risky on skew, which is why Spark prefers sort-merge by default.

3.4 · How Spark chooses — the decision order

Every plan you'll read comes out of this ladder, top to bottom:

1. Is there a hint? Priority: BROADCAST → MERGE → SHUFFLE_HASH → SHUFFLE_REPLICATE_NL.
2. Else, is one side estimated ≤ 10 MB? → Broadcast Hash Join.
3. Else, an equi-join with preferSortMergeJoin = true → Shuffle Sort-Merge Join.
4. Else (small-ish side, preferSortMergeJoin = false) → Shuffle Hash Join.
5. No equi-key → nested-loop / Cartesian (slow — usually a bug).

Figure A · the tipping point

Watch the plan flip at 10 MB

Drag the small side's estimated size. As it crosses the 10 MB threshold the chosen strategy flips Broadcast Hash Join ⇄ Sort-Merge Join, the plan redraws (one BroadcastExchange vs two Exchange nodes), and the "big side shuffled?" readout flips with it. Toggle Force broadcast() to override the estimate. This figure isolates one thing — where the decision boundary is; where the rows physically move is Figure B.

04 · The code — and how to verify it

Three moves, each with its proof

Every technique below is paired with how to prove it worked — the plan node or Spark-UI signal. Apply, then verify; never assume.

Force (or guarantee) a broadcast join

pythonfrom pyspark.sql.functions import broadcast

# orders is huge; country is tiny. broadcast() tells the planner:
# ship `country` to every executor and DO NOT shuffle `orders`.
joined = orders.join(broadcast(country), on="country_code", how="inner")

# VERIFY — .explain(mode="formatted") must show a BroadcastHashJoin, a
# BroadcastExchange on the country side, and NO Exchange on the orders side:
joined.explain(mode="formatted")
#   *(n) BroadcastHashJoin ... (orders.country_code = country.country_code)
#   :- *(n) <orders scan, no Exchange>          <-- big side NOT shuffled  ✅
#   +- BroadcastExchange ...                    <-- small side broadcast    ✅

Equivalent Spark SQL hint — same effect, same plan node:

sql-- The hint goes immediately after SELECT. BROADCAST(country) == broadcast(country).
SELECT /*+ BROADCAST(country) */ o.*, c.country_name
FROM   orders o
JOIN   country c ON o.country_code = c.country_code;
-- VERIFY: same BroadcastHashJoin + BroadcastExchange in the Spark UI SQL tab.

Tune the auto-broadcast threshold, deliberately

python# Raise the threshold so a ~50 MB dimension still auto-broadcasts (OSS default 10 MB).
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 50 * 1024 * 1024)  # bytes

# Or disable auto-broadcast entirely to FORCE a sort-merge join (to A/B a plan):
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)  # -1 = never auto-broadcast

# VERIFY — re-run .explain() and confirm the join node flipped
# (BroadcastHashJoin <-> SortMergeJoin), then reset the conf.

Contrast: naive big-vs-small vs a forced broadcast

python# NAIVE: if Spark mis-estimates `country` as big, it shuffles the 2 TB orders table.
slow = orders.join(country, "country_code")            # may plan SortMergeJoin → double shuffle

# RIGHT: force the broadcast so only the tiny side moves.
fast = orders.join(broadcast(country), "country_code")  # BroadcastHashJoin → no big-side shuffle

# VERIFY — Spark UI → SQL tab: `fast` has NO Exchange on the orders branch;
# `slow` shows two Exchange nodes + Sort + SortMergeJoin. Stages tab: `slow`
# has large Shuffle Read/Write on the orders side; `fast` has none.

Figure B · where the rows go

See it run across the cluster

This figure isolates a different thing from Figure A: not the boundary, but where the data physically lands. Step a broadcast join across 1 driver + 3 executors — watch only the tiny dim cross the wire while the big orders partitions stay put — then flip to the sort-merge join and watch both sides shuffle. The contrast is the whole lesson. Use Prev / Play / Next.

05 · Decision table

Which strategy — at a glance

Read this by the situation you're in — it answers "which plan should I expect, and what's the risk?" without re-explaining the mechanisms (that's §3).

06 · Edge cases & limitations

Where it gets subtle

New nuance an interviewer probes — not a re-run of the three strategies:

• The dimension that's almost under the threshold. A column projection or filter can shrink it enough to broadcast — but pre-AQE Spark may not re-estimate, so force broadcast() or lean on AQE's runtime switch.
• Broadcasting the wrong (big) side. broadcast() is a hint; broadcasting a 5 GB table OOMs the driver / blows spark.driver.maxResultSize (Lesson 03). Only hint the genuinely small side.
• Outer joins limit broadcast. You can broadcast only the side allowed to be the build side (e.g. the right side of a LEFT outer join); the planner may refuse otherwise.
• No equi-key (a.x > b.y): falls back to a nested-loop join — broadcast a tiny side or rewrite the query.

Hard limits worth stating plainly:

• BHJ has no "streamed" broadcast. The small side must fit in driver and executor memory. Above that budget, sort-merge (or bucketing, Lesson 11) is the only safe choice.
• The auto-broadcast decision uses an estimate. Bad or missing stats → wrong choice. Run ANALYZE TABLE … COMPUTE STATISTICS or rely on AQE's runtime sizes.
• OSS vs Databricks thresholds differ. autoBroadcastJoinThreshold is 10 MB in OSS Spark; on Databricks, AQE also applies a higher runtime switch (spark.databricks.adaptive.autoBroadcastJoinThreshold, 30 MB). State which you mean when you quote a number.

07 · Common mistakes

Gotchas — the mistake and its one-line fix

08 · References

Sources

• Apache Spark — SQL Performance Tuning (join strategies, AQE)
• Apache Spark — Join hints (BROADCAST / MERGE / SHUFFLE_HASH / SHUFFLE_REPLICATE_NL)
• Apache Spark — Configuration (autoBroadcastJoinThreshold, shuffle.partitions, preferSortMergeJoin)
• Azure Databricks — Adaptive Query Execution (30 MB runtime broadcast switch)