Spark Join Strategies on Databricks: Broadcast, Sort-Merge, Shuffle Hash, and What AQE Changes

Most slow PySpark jobs I've debugged weren't slow because of bad code. They were slow because Spark picked a join strategy that made sense for the statistics it had, and those statistics were wrong. If you can read a physical plan and know why Spark chose a broadcast hash join over a sort-merge join, you can fix a big share of performance problems in minutes. It's also one of the most common senior data engineering interview topics.

This post walks through the join strategies Spark actually uses, how it chooses between them, how Adaptive Query Execution (AQE) changes that decision at runtime, and the hints you can use when you know better than the optimizer.

The strategies you'll see in a plan

For equi-joins (joins on a.key = b.key), Spark has three main physical strategies:

Strategy	Shuffle?	Sort?	Good when
Broadcast hash join (BHJ)	No (small side is broadcast)	No	One side is small enough to fit in memory on every executor
Shuffle hash join (SHJ)	Yes, both sides	No	Both sides are large, but each partition of the build side fits in memory
Sort-merge join (SMJ)	Yes, both sides	Yes	Both sides are large; the safe default

For non-equi joins (for example a.ts BETWEEN b.start AND b.end), Spark falls back to a broadcast nested loop join or a Cartesian product. Those are the ones that quietly turn a 5-minute job into a 5-hour job.

How to see which one you got

Always check the plan before tuning anything:

from pyspark.sql import functions as F

orders = spark.table("sales.orders")          # large fact table
customers = spark.table("sales.customers")    # dimension table

joined = orders.join(customers, "customer_id", "inner")
joined.explain("formatted")

In the output, look for one of BroadcastHashJoin, SortMergeJoin, or ShuffledHashJoin. A BroadcastExchange node means one side is being shipped to every executor. An Exchange hashpartitioning(customer_id, ...) on both sides means a full shuffle.

With AQE on (the default on Databricks), the plan you see before execution starts with AdaptiveSparkPlan isFinalPlan=false. The final plan is only known after the job runs, so check the SQL tab in the Spark UI to see what actually executed.

How Spark chooses

At planning time, for an equi-join, Spark roughly follows this order:

1. If there's a join hint, honour it (when it's valid for the join type).
2. If one side's estimated size is below spark.sql.autoBroadcastJoinThreshold (10 MB by default) and that side can be the build side for this join type, use a broadcast hash join.
3. If spark.sql.join.preferSortMergeJoin is false and the build side is small enough per partition, consider a shuffle hash join.
4. Otherwise, use a sort-merge join (when the keys are sortable).

The key word in step 2 is estimated. Spark estimates size from table statistics or file sizes, and after filters, aggregations, or UDFs those estimates can be badly off in either direction. That's why AQE matters.

Which side can be broadcast

This is a classic interview trap. The broadcast side is the build side, and not every join type allows every side to be the build side:

• inner: either side can be broadcast.
• left, left_semi, left_anti: only the right side can be broadcast.
• right: only the left side can be broadcast.
• full_outer: a broadcast hash join isn't possible.

So if you write small_df.join(big_df, "id", "left") and expect the small table to be broadcast, it won't be. Spark has to keep every row of the left side, so it can't build a hash table from it. Flip the join around or change the join type.

AQE: fixing the plan at runtime

With spark.sql.adaptive.enabled=true, Spark re-plans after each shuffle stage finishes, using real sizes instead of estimates. For joins, two things matter most:

Sort-merge join to broadcast join. If the actual shuffled size of one side ends up below spark.sql.adaptive.autoBroadcastJoinThreshold (which defaults to the same value as spark.sql.autoBroadcastJoinThreshold), AQE switches the join to a broadcast hash join. The shuffle already happened, but AQE can use a local shuffle reader to avoid a second network read and skip the sort.

Sort-merge join to shuffle hash join. If every post-shuffle partition is smaller than spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold, AQE can use a shuffle hash join instead, skipping the sort. That setting defaults to 0 (off) in open-source Spark, so it only kicks in if you set it.

A typical pattern: a filter cuts a 2 GB dimension down to 4 MB, the optimizer can't tell, plans a sort-merge join, and AQE saves you by converting it to a broadcast join at runtime.

spark.conf.set("spark.sql.adaptive.enabled", "true")

active = (spark.table("sales.customers")
          .filter(F.col("status") == "ACTIVE")
          .filter(F.col("country") == "CA"))

result = spark.table("sales.orders").join(active, "customer_id")
result.write.mode("overwrite").saveAsTable("sales.orders_active_ca")
# Spark UI > SQL tab: the final plan may show BroadcastHashJoin
# even though the initial plan showed SortMergeJoin.

Join hints: when you know better

Hints override the size-based choice. In PySpark:

from pyspark.sql.functions import broadcast

# Force a broadcast of the dimension table
orders.join(broadcast(customers), "customer_id")

# Same thing with the generic hint API
orders.join(customers.hint("broadcast"), "customer_id")

# Ask for a shuffle hash join or a sort-merge join explicitly
orders.join(customers.hint("shuffle_hash"), "customer_id")
orders.join(customers.hint("merge"), "customer_id")

And in SQL:

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

If you put conflicting hints on both sides, Spark prioritises them in this order: BROADCAST, then MERGE, then SHUFFLE_HASH, then SHUFFLE_REPLICATE_NL. Hints that aren't valid for the join type (for example broadcasting the left side of a left outer join) are ignored, and Spark logs a warning.

When broadcasting goes wrong

Broadcasting isn't free. The driver collects the small side, builds the hash relation, and sends it to every executor. Common failure modes:

• Driver out of memory. A "small" table that's actually 1.5 GB after decompression can take down the driver. Raising the threshold to something like 1 GB because it made one job faster is a classic way to break another job.
• Broadcast timeout. If building the broadcast side involves heavy upstream work, you can hit spark.sql.broadcastTimeout (300 seconds by default). Caching or materialising the small side first usually helps more than raising the timeout.
• Hard size limit. Spark refuses to broadcast tables over 8 GB, regardless of hints.
• Executor memory pressure. Every executor holds a copy. A 500 MB broadcast across 200 executors is 100 GB of cluster memory.

A safe pattern is to compute the small side explicitly and check its size before forcing the hint:

dim = spark.table("sales.customers").select("customer_id", "segment")
dim_bytes = (spark.sql("DESCRIBE DETAIL sales.customers")
             .select("sizeInBytes").first()[0])  # Delta table size on disk

if dim_bytes < 200 * 1024 * 1024:   # 200 MB on disk, adjust for your cluster
    joined = orders.join(broadcast(dim), "customer_id")
else:
    joined = orders.join(dim, "customer_id")

Size on disk is compressed, so leave plenty of headroom; Parquet often expands several times in memory.

Keeping statistics honest

The optimizer can only choose well if it knows table sizes. On Databricks, run this on dimension tables that feed important joins:

ANALYZE TABLE sales.customers COMPUTE STATISTICS FOR ALL COLUMNS;

Column statistics also help the cost-based optimizer (spark.sql.cbo.enabled) reorder multi-way joins so the most selective joins happen first.

Interview questions worth practising

1. Why can't Spark broadcast the left side of a left outer join?
2. Your job shows SortMergeJoin in explain() but BroadcastHashJoin in the Spark UI. What happened?
3. When would you prefer a shuffle hash join to a sort-merge join?
4. Setting autoBroadcastJoinThreshold to 2 GB made one job faster and broke another. Why?
5. A join on BETWEEN conditions is very slow. What's Spark doing, and how would you rewrite it? (Hint: bucket the range into discrete keys so it becomes an equi-join, then filter.)

If you can answer those clearly, with a plan in front of you, you'll handle most join questions in a Databricks interview.

---

I've put together a free 10-question sample of Azure Databricks and PySpark interview questions with worked answers, if you'd like more practice like this: free sample. The full set is in the Azure Databricks & PySpark Interview Pack.