Joins in Spark

Analogy categoryAnalogies

Join Strategies Spark Chooses From

Spark's Catalyst planner picks between three main physical join strategies based on table sizes and available statistics: broadcast hash join, shuffle hash join, and sort-merge join. The choice matters enormously for performance, since a poorly chosen strategy can turn a join that should take seconds into one that shuffles terabytes across the network. You can inspect which strategy was chosen with .explain(), and override it with join hints when Spark's cost-based optimizer picks wrong due to missing or stale table statistics.

🏏

Cricket analogy: It's like choosing a bowling strategy based on the pitch report — a spin-friendly pitch calls for two spinners, a green top calls for pace, and reading it wrong before the toss costs the match regardless of talent.

Broadcast Hash Join

When one side of a join is small enough to fit comfortably in each executor's memory, below spark.sql.autoBroadcastJoinThreshold which defaults to 10MB, Spark serializes that entire table and sends a full copy to every executor. Each executor then performs a local hash join against its partition of the large table with zero network shuffle of the large side, making broadcast joins dramatically faster than shuffle-based alternatives whenever the size asymmetry holds. You can force this strategy with broadcast(df) when Spark's automatic detection misses it due to inaccurate size statistics.

🏏

Cricket analogy: It's like handing every fielder a laminated card of the batsman's scoring zones before the over instead of radioing instructions ball by ball — one small, cheap distribution to everyone avoids constant back-and-forth communication during play.

python

from pyspark.sql.functions import broadcast

orders = spark.read.parquet("s3://data/orders.parquet")   # ~500 GB
regions = spark.read.parquet("s3://data/regions.parquet")  # ~2 MB

# Force a broadcast hash join: 'regions' is small enough to
# ship to every executor, avoiding a shuffle of 'orders'
joined = orders.join(broadcast(regions), "region_id")

# Two large tables: Spark defaults to a sort-merge join
customers = spark.read.parquet("s3://data/customers.parquet")  # ~200 GB
enriched = orders.join(customers, "customer_id")

joined.explain(True)  # inspect which physical join strategy was chosen

Sort-Merge Join

When both sides of a join are too large to broadcast, Spark defaults to a sort-merge join: it shuffles both DataFrames so that rows with the same join key land on the same partition, sorts each partition by the key, and then merges the two sorted streams in a single linear pass. This strategy scales to arbitrarily large tables because it never needs to hold an entire side in memory at once, but it pays the full cost of shuffling both datasets across the network plus the sort, making it noticeably slower than a broadcast join when one side happens to be small.

🏏

Cricket analogy: It's like reconciling two separate scorebooks kept by different scorers — both books get sorted by over number first, then merged line by line, which works for any match length but takes real time compared to just trusting one pre-verified copy.

spark.sql.autoBroadcastJoinThreshold (default 10MB) controls the automatic broadcast join cutoff — raise it if you have executors with plenty of memory and know a slightly larger dimension table is safe to broadcast, or set it to -1 to disable automatic broadcasting entirely for debugging.

Skewed Joins and Adaptive Query Execution

A join key with extreme skew, for example a 'country' column where 90% of rows are 'US', causes the sort-merge join's US partition to become a massive straggler task that dwarfs every other partition and can stall the whole stage for many minutes while other tasks finish in seconds. Adaptive Query Execution's skew join optimization, enabled by default in modern Spark versions, detects oversized partitions at runtime from shuffle statistics and automatically splits them into smaller sub-partitions that are joined independently, achieving what manual key-salting used to require developers to hand-code.

🏏

Cricket analogy: It's like a net-bowling session where one bowler gets assigned 90% of the deliveries because the roster was drawn up carelessly — that one bowler's arm gives out long before the others have even finished their spell, stalling the whole session.

A heavily skewed join key can silently turn a job that should take minutes into one that hangs for hours, with the Spark UI showing every task finished except one. Always check for stragglers in the final stage before assuming a slow join is simply a matter of adding more executors.

Spark chooses between broadcast hash join, shuffle hash join, and sort-merge join based on size and statistics.
Broadcast hash join ships a small table to every executor, avoiding a shuffle of the large table.
Sort-merge join shuffles and sorts both sides, then merges them, scaling to arbitrarily large tables.
Force a broadcast join with broadcast(df) when automatic detection misses it.
Skewed keys create straggler tasks that can stall an entire stage.
Adaptive Query Execution can split skewed partitions and switch join strategies at runtime.