Large-Scale Joins
Matching keys across partitions without a full cartesian
Joins at scale come in strategy flavors: broadcast (tiny side copied everywhere), shuffled hash (sort-merge/broadcast-like on both sides), and skewed-key treatments. Choosing the strategy is the difference between a fast job and a full-cluster shuffle.
What you will learn
- Compare broadcast, shuffled hash and sort-merge joins
- Pick a strategy from table sizes and key distribution
- Handle skewed-join keys at scale
What is this?
Joins in distributed engines are about collocating the matching rows across partitions. Three families dominate: broadcast (copy the small side to every executor — no shuffle), shuffled hash or sort-merge (both sides shuffled on the join key, then hashed/sorted and merged), and specialty patterns (asymmetric, interval, skewed-key) when features of the data break the standard approach.
Why do you need it?
- The join is where 80% of performance incidents live; strategy selection decides 10x differences.
- A broadcast join turns a shuffle-heavy monster into a no-shuffle microsecond job.
- Skew makes the 'standard' shuffle join pathological without a salting/adaptive intervention.
- Knowing strategy mechanics lets you interpret physical plans (BroadcastHashJoin vs SortMergeJoin).
Mental model
Getting phone numbers for 100,000 customers
Join strategies
Broadcast: photocopy the small phone book to every branch so each outlet looks it up instantly. Shuffled join: every branch mails its customer requests to a national office by area code and the area offices compare them. Same result, wildly different logistics.
How it works
- Broadcast join: the smaller side is collected to the driver, serialized, and sent to every executor; each executor builds a hash table and probes with its local data — zero shuffles.
- Shuffled hash join: both sides are repartitioned by join key; matching keys land on the same executor, which builds a hash table on one side and probes the other.
- Sort-merge join: both sides are range-partitioned by key; each executor sorts and walks sorted runs together — used when memory is tight (large keys).
- Skewed join handling: salting hot keys, splitting NULLs, or AQE skewJoin auto-splitting.
- The optimizer (Spark's) chooses a strategy via stats (metrics per side) and thresholds (broadcast join size limit).
Broadcast vs shuffled join.
In production
- A 10MB country dim joined to a 50GB fact uses broadcast — no shuffle at all.
- Two large event tables join via sort-merge when memory per partition can't hold one side.
- A pathological NULL-keys join gets salted so unmatched rows spread over many reducers.
Best practices
- Broadcast any small stable dimension (< ~10MB default threshold) — it's nearly free.
- Prefer the engine's automatic choice unless you have data (stats) to distrust it.
- Prefilter and aggregate each side *before* the join to shrink the shuffle.
- Keep bucket/partition keys aligned between fact and dimension for collocated joins.
Common mistakes
- Assuming joins always shuffle — and missing a free broadcast because a warning was ignored.
- Joining on a non-unique key with no dedup and silently multiplying rows.
- Building a cartesian product with a skewed key and worshipping the shuffle.
- Mistaking cost-based defaults: forcing broadcast with a huge table OOMs the driver.
Common issues
Driver OOM during joinA forced broadcast join on a table larger than the driver memory
Raise the broadcast threshold carefully, or let Spark shuffle-join instead.
Join output row count explodes vs inputJoin key isn't unique and doesn't dedup (fatal grain mistake)
Aggregate/dedup each side to grain before joining.
Sort-merge join freezes on one partitionSkewed key range distribution concentrates comparisons
Range-forced keys, salting, or AQE's adaptive skew split.
Debugging tips
- If the plan shows SortMergeJoin and no Broadcast, ask why (side too big? stats missing?).
- Rows multiply suspiciously (fact x dim)? Check the join key for collisions or nulls.
- Skewed (straggler reducers)? Salt keys or run both sides through a pre-aggregate.
- Compare join strategies by explain() both ways — the physical plan is the truth.
Interview checkpoint
Hands-on lab
- Run a broadcast join on a small dim and confirm zero shuffle
- Force a shuffled join to compare plan and timing
- Reproduce a skewed join straggler with a hot key
Compare Broadcast vs Shuffled vs Skewed Joins
35 min · advanced
Quick quiz
A 5MB country-dim joined to a 1TB fact should use which strategy?
A join on a non-unique key that collides can silently multiply your output rows.