What's the best way to optimize a slow Spark join?

Summary

• Most slow Spark joins are shuffle problems, so the first fix is to avoid shuffling the large side: broadcast the small side, or store both sides pre-partitioned on the join key with bucketing or a storage partition join.
• Keep Adaptive Query Execution (AQE) on, as it is by default since Spark 3.2.0. At runtime it can turn a sort-merge join into a broadcast hash join and split skewed partitions.
• Skewed keys need their own treatment: AQE's skew join splits oversized partitions in sort-merge joins automatically, and salting the hot keys is the manual last resort.
• Shrink inputs before the join and give the optimizer statistics: filter and project early, let dynamic partition pruning and runtime Bloom filters cut the large side, and run ANALYZE TABLE on join keys.

What's the best way to optimize a slow Spark join?

Move less data. A join is slow when Spark shuffles large inputs across the cluster or when one task gets most of the rows, so the best technique is the one that removes the shuffle or evens out the work. Start with the cheapest change that does that: a broadcast, early filters or AQE's defaults.

Join optimization techniques at a glance

Technique	Use it when	Trade-off
Broadcast join	One side is small enough to copy to every executor	The whole side is sent to every worker node
AQE	Always; it's on by default	A runtime conversion isn't as efficient as planning the broadcast up front
AQE skew join	A few keys hold most of the rows	Applies to sort-merge joins
Salting	Skew that AQE doesn't split	Requires code changes and replicates the other side
Bucketing	Large tables joined repeatedly on the same key	Bucket count is fixed at write time; file-based tables only
Storage partition join	Both sides are compatible V2 tables, such as Iceberg	Needs matching partition transforms and several configs
Filter and project early	The join reads rows or columns it later drops	None
Statistics and hints	The optimizer picks the wrong strategy	A hint keeps forcing its strategy as data grows

Broadcast the small side

Spark broadcasts a side automatically when its estimated size is below spark.sql.autoBroadcastJoinThreshold (default 10 MB; -1 disables it). Each executor joins its partitions of the large side against a local copy, so the large side never shuffles. When the estimate is wrong, mark the side with broadcast() or a BROADCAST hint, which Spark honors regardless of the threshold. Keep an equality condition in the join: without an equi-join key, a broadcast becomes a broadcast nested loop join.

from pyspark.sql.functions import broadcast

orders = spark.table("sales.orders").where("order_date >= '2026-01-01'")
stores = spark.table("sales.stores").select("store_id", "region")

joined = orders.join(broadcast(stores), "store_id")
joined.explain(mode="formatted")  # expect BroadcastHashJoin

Let AQE and skew handling fix the plan

AQE (spark.sql.adaptive.enabled) re-optimizes with runtime statistics. It converts a sort-merge join to a broadcast hash join when a side's runtime size falls below spark.sql.adaptive.autoBroadcastJoinThreshold, which defaults to the static threshold.
For skew, AQE splits (and replicates if needed) skewed partitions in sort-merge joins. A partition counts as skewed when it's larger than spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5.0) times the median partition size and also larger than spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (default 256MB). Lower those when hot keys slip through. If AQE still can't even out the tasks, salt the key: append random integers as suffixes to the skewed values, replicate the other side once per suffix and join on the salted key.

SET spark.sql.adaptive.skewJoin.skewedPartitionFactor=3;
SET spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=128MB;
ANALYZE TABLE sales.orders COMPUTE STATISTICS FOR COLUMNS customer_id;
EXPLAIN COST
SELECT o.order_id, c.segment
FROM sales.orders o JOIN sales.customers c ON o.customer_id = c.customer_id;

Pre-partition the tables you join often

Bucketing writes a table into a fixed number of buckets by a column, using bucketBy() with saveAsTable() or CLUSTERED BY ... INTO n BUCKETS, so later joins on that key can avoid the shuffle. Bucket both sides on the join key with the same count; spark.sql.bucketing.coalesceBucketsInJoin.enabled (default false) handles counts where one divides the other. For V2 sources such as Iceberg, the storage partition join generalizes bucket joins to tables partitioned by functions like bucket(8, id). With spark.sql.sources.v2.bucketing.enabled (default true) and the related settings, the plan has no Exchange nodes before the join.

Filter early and feed the optimizer

Select only the columns you need and filter before the join. Spark helps with Parquet filter pushdown (spark.sql.parquet.filterPushdown), dynamic partition pruning (spark.sql.optimizer.dynamicPartitionPruning.enabled), which generates a partition predicate when a partition column is a join key, and runtime Bloom filters (spark.sql.optimizer.runtime.bloomFilter.enabled), which shrink shuffle data when one side has a selective predicate. All three default to true. Cost-based optimization and join reordering (spark.sql.cbo.enabled, spark.sql.cbo.joinReorder.enabled) are off by default and rely on the statistics ANALYZE TABLE collects.

How Databricks approaches slow joins

Databricks runs these Spark strategies with AQE on by default and adds its own. Photon, the Databricks-native vectorized query engine, replaces sort-merge joins with hash joins and uses a columnar shuffle. AQE on Databricks handles skew in both sort-merge and shuffle hash joins, uses its own runtime broadcast threshold (spark.databricks.adaptive.autoBroadcastJoinThreshold, default 30MB), and sizes shuffles automatically when spark.sql.shuffle.partitions is set to auto. Range join optimization covers point-in-interval and interval-overlap joins, and Databricks SQL applies it automatically. For layout, liquid clustering on the columns you filter by improves data skipping.

FAQs

Should I just raise spark.sql.autoBroadcastJoinThreshold?

Only for tables you know will stay small. Every worker node receives a full copy of the broadcast side, and a BROADCAST hint keeps copying it even after the table outgrows the threshold.

Why didn't my BROADCAST hint take effect?

Spark isn't guaranteed to use a hinted strategy, because a given strategy may not support every join type. Check the physical plan with explain() and look for BroadcastHashJoin.

Does AQE make join hints unnecessary?

Often. AQE can switch to a broadcast at runtime, but the Spark docs note that this isn't as efficient as planning the broadcast in the first place.