What is the best way to optimize a slow Spark join?

Summary

• The best way to optimize a slow Spark join is to diagnose it first: find the join in the Spark UI SQL tab, read the physical join strategy, then check sizes, skew, spill and output rows before changing anything.
• The strategy name tells you most of the story. BroadcastHashJoin avoids shuffling the large side, SortMergeJoin and ShuffledHashJoin shuffle both sides, and BroadcastNestedLoopJoin or CartesianProduct usually mean the join has no equi-join key.
• A sort-merge join on a table that is actually small usually points to missing or stale statistics. EXPLAIN COST shows the estimates, and ANALYZE TABLE ... COMPUTE STATISTICS or a BROADCAST hint fixes the choice.
• Skew shows up as one task with far more shuffle read than its peers, spill as the spill size metric on the join's sort operators, and many-to-many keys as join output that dwarfs its inputs.

What is the best way to optimize a slow Spark join?

Diagnose before you tune. A slow join has one of a handful of causes (the wrong strategy, a bad size estimate, a hot key, spill or row explosion) and each needs a different fix. The walkthrough below uses only what open-source Apache Spark gives you: the Spark UI's SQL tab, EXPLAIN, and table statistics.

Step 1: Find the join in the SQL tab

Every DataFrame action and SQL query appears in the SQL tab. Open the slow query's execution detail page: its plan graph shows each operator with metrics inline, the Details link expands the logical and physical plans, and the header lists the query's jobs and stages so you can jump to the join's stage.

Step 2: Identify the join strategy

Before a run, ask for the plan; after a run, read the final plan in the SQL tab, because Adaptive Query Execution (AQE) can change the strategy mid-query. The root reads AdaptiveSparkPlan isFinalPlan=true once it has finished.

joined = orders.join(customers, "customer_id")
joined.explain(mode="formatted")  # physical plan outline plus node details
joined.explain(mode="cost")       # logical plan with size and row estimates

Spark's physical join operators are BroadcastHashJoin, ShuffledHashJoin, SortMergeJoin, BroadcastNestedLoopJoin and CartesianProduct. An Exchange hashpartitioning under both sides of the join means both inputs are being shuffled.

Step 3: Compare estimates with reality

Spark broadcasts a side automatically when its estimated size is below spark.sql.autoBroadcastJoinThreshold (default 10 MB). The Spark docs warn that missing or inaccurate statistics hinder plan selection. Check three places: DESCRIBE EXTENDED for table statistics, EXPLAIN COST for the optimizer's estimates, and the data size metric on each exchange in the SQL tab for what actually moved. Runtime statistics appear in the plan details as Statistics(..., isRuntime=true).

Step 4: Look for skew, spill and row explosion

• Skew. Open the join's stage and compare the slowest task with the rest in the summary metrics for Duration and Shuffle Read Size / Records. One task reading far more than its peers is a hot key. If AQE split it, the final plan shows SortMergeJoin(skew=true).
• Spill. A sort-merge join sorts each side, and the spill size metric on those Sort operators counts bytes spilled from memory to disk. The stage page reports spill too, as Shuffle spill (memory) and Shuffle spill (disk).
• Row explosion. Compare number of output rows on the join with its inputs. Output far larger than either input means duplicate keys on both sides are multiplying rows.

Step 5: Pick the fix that matches the finding

What you found	Likely cause	Fix
SortMergeJoin where one side is small	Missing statistics, or an estimate above the broadcast threshold	Run ANALYZE TABLE ... COMPUTE STATISTICS, raise spark.sql.autoBroadcastJoinThreshold, or add a BROADCAST hint
BroadcastNestedLoopJoin or CartesianProduct	No equi-join key in the condition	Rewrite the condition to include an equality on a key, if the logic allows it
One task with an outsized shuffle read	A hot key	Check AQE's skew-join thresholds, or broadcast the small side, which removes the shuffle that concentrates the key
Large spill size on the join's sorts	Partitions too big for task memory	Raise spark.sql.shuffle.partitions (or spark.sql.adaptive.coalescePartitions.initialPartitionNum when AQE coalesces), or give tasks more memory
Output rows far above input rows	Many-to-many keys	Deduplicate or aggregate one side before joining, or fix the key
Big exchanges on both sides every run	Both inputs shuffled each time	Filter rows and drop unused columns first; on compatible V2 data sources, Storage Partition Join uses the existing layout to avoid the shuffle

For the statistics fix and an explicit broadcast:

ANALYZE TABLE customers COMPUTE STATISTICS FOR ALL COLUMNS;

SELECT /*+ BROADCAST(c) */ o.*, c.segment
FROM orders o JOIN customers c ON o.customer_id = c.customer_id;

Then re-run and repeat Steps 2 to 4 on the new final plan. The fix is done when the plan and metrics say so.

How Databricks approaches slow joins

The same diagnosis works on Databricks, with extra tooling. The query profile shows each operator with time spent, rows processed and memory, and Databricks lists exploding joins among the common mistakes it helps you find. Its Top operators panel ranks the longest-running operators. For Unity Catalog managed tables, predictive optimization runs ANALYZE and collects statistics when data is written, which keeps join-size estimates current. AQE is on by default. See Query profile.

FAQs

Should I just increase spark.sql.shuffle.partitions?

Only when spill shows that partitions are too large. More partitions don't help a hot key, because hash partitioning still sends every row for that key to one partition, and they don't fix a missed broadcast.

Why did Spark ignore my BROADCAST hint?

The Spark docs say there's no guarantee Spark will use the hinted strategy, because a given strategy may not support all join types. Check the final plan to see what ran.

How do I stop Spark from broadcasting a table?

Set spark.sql.autoBroadcastJoinThreshold to -1, which disables automatic broadcasting. AQE's runtime threshold, spark.sql.adaptive.autoBroadcastJoinThreshold, defaults to the same value, so it follows unless you've set it separately.