Spark Join Strategies: Broadcast Join vs Sort-Merge Join vs Shuffle Hash Join

Last Updated: July 2026 | 14 min read

Quick Answer: Spark has three main join strategies. A broadcast hash join copies a small table to every executor and joins with no shuffle — fastest, but one side must fit in memory. A sort-merge join shuffles and sorts both tables on the join key — Spark's default when both sides are large. A shuffle hash join shuffles both sides and builds an in-memory hash table on the smaller one — a middle ground Spark uses only when you disable the sort-merge preference. Spark picks automatically based on table size and the spark.sql.autoBroadcastJoinThreshold (default 10 MB), and you can override it with a broadcast() hint.

Joins are where most Spark jobs live or die. The same query can finish in 30 seconds or run for an hour depending on which Spark join strategy the optimizer picks — and the difference almost always comes down to one thing: the shuffle. Get the join strategy right and you skip moving terabytes across the network; get it wrong and you shuffle data you never needed to, or crash the driver trying to broadcast a table that was never small.

This guide explains the three strategies, shows exactly how Spark chooses between them, how to force a broadcast join (and why yours might not be triggering), and how Adaptive Query Execution changes the decision at runtime. All examples are PySpark; the same strategies and hints apply in Scala and Spark SQL.

Why Spark has more than one join strategy

A join strategy is the physical algorithm Spark uses to match rows from two tables on a key — and Spark has several because there is no single best way to join at scale. The right choice depends entirely on how big each side is and whether it can move without a shuffle.

Every distributed join faces the same problem: rows that share a key must end up on the same machine. There are only two ways to make that happen:

1. Move the small side everywhere — copy one table in full to every executor so each machine has what it needs locally. This is a broadcast.
2. Move both sides by key — repartition both tables so matching keys land on the same executor. This is a shuffle, the single most expensive operation in Spark.

Broadcasting is cheap but only works when one side is small. Shuffling works for any size but is expensive. Every Spark join strategy is a different point on that trade-off.

The three Spark join strategies compared

Here is how the main strategies stack up. (Spark also has two fallbacks for non-equi joins — broadcast nested loop join and cartesian product — covered at the end.)

The rule of thumb: broadcast when you can, sort-merge when you must, shuffle-hash rarely.

Broadcast hash join — the one you want

A broadcast hash join sends a full copy of the small table to every executor, builds a hash table from it in memory, and streams the large table past it — with zero shuffle of the large side. It is almost always the fastest join when it applies.

Spark plans it automatically when one side's estimated size is under spark.sql.autoBroadcastJoinThreshold (default 10 MB). Because nothing large moves across the network, it sidesteps the shuffle entirely — which is why forcing a broadcast is the number-one fix for a slow join.

from pyspark.sql.functions import broadcast

# orders is huge, countries is a tiny lookup table
result = orders.join(broadcast(countries), "country_code")

The catch is memory. The small side is collected to the driver, then shipped to every executor and held in memory. Broadcast a table that is not actually small and you get a driver OutOfMemoryError or a broadcast timeout. Broadcasting is a scalpel, not a hammer.

Sort-merge join — the reliable default

A sort-merge join shuffles both tables so matching keys share a partition, sorts each partition by the join key, then walks the two sorted streams together to emit matches. It is Spark's default for two large tables because it is the most robust: it scales to any size and spills to disk when memory is tight.

The price is the full shuffle and a sort on both sides. For a join between two multi-terabyte tables that is unavoidable — but if one side turns out to be small, sort-merge is pure waste, which is exactly what a broadcast hint (or AQE) fixes.

# Both large → Spark plans a sort-merge join automatically
big_orders.join(big_customers, "customer_id")

Shuffle hash join — the rare middle ground

A shuffle hash join shuffles both sides on the key like sort-merge, but instead of sorting it builds an in-memory hash table on the smaller partition and probes it with the larger — trading the sort cost for memory pressure. It fits the middle case: a side too big to broadcast but small enough to hash per partition.

Spark keeps it off by default (spark.sql.join.preferSortMergeJoin=true) because a hash table that does not fit in memory triggers an executor OOM, whereas sort-merge safely spills. Enable it deliberately with a hint when you know the build side is comfortably small and the keys are costly to sort.

How Spark chooses a join strategy

For an equi-join, Spark walks a fixed priority order and picks the first strategy that applies: broadcast hash join → shuffle hash join → sort-merge join. Understanding this order is what lets you predict — and control — the plan.

1. Broadcast hash join — if either side has a broadcast hint, or its estimated size is below spark.sql.autoBroadcastJoinThreshold.
2. Shuffle hash join — if spark.sql.join.preferSortMergeJoin is false and one side is small enough to build a local hash map.
3. Sort-merge join — if the join keys are sortable. This is the default catch-all.
4. Broadcast nested loop / cartesian — the fallbacks for non-equi joins (>, <, BETWEEN, or no key).

You can see which one Spark chose with .explain() — look for BroadcastHashJoin, SortMergeJoin, or ShuffleHashJoin in the physical plan:

result.explain()
# == Physical Plan ==
# *(2) BroadcastHashJoin [country_code#12], [country_code#31], Inner, BuildRight
#   ...

Reading the physical plan is the single most useful Spark-tuning skill — it tells you the truth about what will run, before you wait an hour to find out. If you are still building fluency with the DataFrame API and expressions, our free interactive Spark with Scala course lets you run every join and function live in the browser, and the PySpark DataFrame fundamentals guide covers the basics in Python.

How to force (or block) each join strategy

Spark's size estimates are often wrong — stale statistics, post-filter sizes, compressed files — so overriding the planner with join hints is routine tuning, not a hack. Every strategy has a hint.

from pyspark.sql.functions import broadcast

# Force broadcast (DataFrame API)
orders.join(broadcast(dim_country), "country_code")

-- Force each strategy with a Spark SQL hint
SELECT /*+ BROADCAST(c) */      * FROM orders o JOIN country  c ON o.cc = c.cc;
SELECT /*+ SHUFFLE_HASH(c) */   * FROM orders o JOIN country  c ON o.cc = c.cc;
SELECT /*+ MERGE(c) */          * FROM orders o JOIN big_dim  c ON o.cc = c.cc;   -- sort-merge
SELECT /*+ SHUFFLE_REPLICATE_NL(c) */ * FROM a JOIN b ON a.x < b.y;               -- non-equi

The key configs to know:

To make a large lookup broadcast, raise the threshold (e.g. spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 50*1024*1024)) rather than hinting blindly — but measure the in-memory size first.

Why your broadcast join isn't triggering

If you expected a broadcast join and got a sort-merge join, it is almost always one of six causes. This is the most common real-world Spark join question, so here is the full checklist:

• The estimated size is over the threshold. Spark broadcasts based on estimated size, not the row count you have in your head. A "small" table can estimate large.
• Statistics are stale. Without table stats, Spark over-estimates. Run ANALYZE TABLE t COMPUTE STATISTICS or rely on AQE, which uses real runtime sizes.
• In-memory size ≫ file size. A 10 MB Parquet file can expand to 150 MB+ in memory once decompressed and deserialized — over the threshold.
• Auto-broadcast is disabled. Someone set spark.sql.autoBroadcastJoinThreshold=-1 (often to stop accidental broadcasts) and forgot.
• The join type forbids it. You cannot broadcast the streamed side of an outer join: for a LEFT OUTER join only the right side can be broadcast, for RIGHT OUTER only the left, and a FULL OUTER join can broadcast neither.
• The side genuinely is too big. Broadcasting it would OOM the driver — Spark is protecting you. Fix the data model instead (pre-aggregate the dimension, or shuffle).

At SolutionGigs, the most common "slow join" ticket we see is a dimension table that everyone believes is small but estimates at 400 MB because of stale stats — a one-line broadcast() hint cuts the job from 40 minutes to under 2.

AQE: how Spark re-plans joins at runtime

Adaptive Query Execution (AQE), on by default since Spark 3.2, fixes the optimizer's biggest weakness — it plans joins with compile-time size guesses, then AQE corrects them using real statistics measured after each shuffle stage. For joins it does three things:

• Dynamically switches to broadcast. If a shuffle reveals one side is actually smaller than spark.sql.adaptive.autoBroadcastJoinThreshold, AQE converts a planned sort-merge join into a broadcast join — catching the small tables the compile-time estimate missed (e.g. a table that only becomes small after a filter).
• Splits skewed joins. AQE detects a partition far larger than the median and splits that one hot key into sub-partitions, so a single straggler task no longer stalls the stage. (See our deep dive on Spark data skew for the manual salting technique AQE automates.)
• Coalesces shuffle partitions. It merges the many tiny partitions a shuffle produces into right-sized ones — related to the repartition vs coalesce trade-off.

AQE does not make join hints obsolete — a broadcast() hint still forces the choice immediately without waiting for a shuffle — but it dramatically reduces how often the planner gets a join wrong.

Common mistakes that ruin Spark joins

• Broadcasting a table that isn't small. The fastest way to OOM the driver. Always confirm the in-memory size, not the file size.
• Ignoring skew. A perfect broadcast join still stalls if one key holds 80% of the rows. Broadcast fixes shuffle cost, not skew — enable AQE skew join or salt the key.
• Leaving autoBroadcastJoinThreshold at 10 MB for a big cluster. On a cluster with 64 GB executors, broadcasting a 100 MB dimension is often a win. Tune the threshold to your hardware.
• Joining on a non-deterministic or type-mismatched key. A bigint↔string key silently casts and can wreck partitioning and matches.
• Not reading .explain(). Every join-tuning session starts by confirming the physical plan — guessing wastes hours. Pair this with sane file layout and partitioning so the join reads only what it needs, and watch for the small files problem that inflates shuffle overhead.

Choosing the right join strategy: a quick decision guide

1. Is one side small (a dimension, lookup, or config table)? → Broadcast hash join. Add broadcast() if Spark doesn't do it automatically.
2. Are both sides large? → Sort-merge join (the default). Make sure the join key is well-distributed to avoid skew, and keep AQE on.
3. Is one side mid-sized, too big to broadcast but small per partition, with keys costly to sort? → Consider a shuffle hash join via SHUFFLE_HASH — but benchmark against sort-merge first.
4. Is it a non-equi join (<, BETWEEN, ranges)? → Spark falls back to broadcast nested loop (if a side is small) or cartesian product. Keep the broadcastable side tiny; these are O(n×m).

Whatever you choose, verify it with .explain() and watch the Spark UI's SQL tab for the actual shuffle sizes — the plan Spark ran is the only source of truth.

Frequently Asked Questions

What is the difference between a broadcast join and a sort-merge join in Spark?

A broadcast hash join copies the small table to every executor and joins locally, avoiding a shuffle entirely — fast, but one side must fit in memory. A sort-merge join shuffles both tables on the join key, sorts each partition, and merges them; it handles two large tables and every join type but pays the full shuffle plus sort cost. Broadcast when one side is small; sort-merge when both are large.

How do I force a broadcast join in Spark?

Wrap the small DataFrame in broadcast(): from pyspark.sql.functions import broadcast, then large.join(broadcast(small), "id"). In Spark SQL, use the /*+ BROADCAST(small) */ hint. This broadcasts that side regardless of autoBroadcastJoinThreshold, as long as it fits in memory. Only hint a table you know is genuinely small — forcing a large broadcast causes a driver out-of-memory error.

Why is my Spark broadcast join not working?

Usually one of six causes: the estimated size exceeds spark.sql.autoBroadcastJoinThreshold (default 10 MB); the threshold is set to -1 (disabled); statistics are stale so Spark over-estimates; the in-memory size is far larger than the file size; the side is genuinely too big; or the join type forbids it — you cannot broadcast the streamed side of an outer join. Add an explicit broadcast() hint or enable AQE.

What is the default join strategy in Spark?

Sort-merge join is the default for equi-joins between two tables that are both too large to broadcast. Spark chooses it because it scales to any size, spills to disk under memory pressure, and supports all join types. Spark only picks a broadcast hash join first when one side is small enough, and only uses a shuffle hash join when you set spark.sql.join.preferSortMergeJoin=false.

What is spark.sql.autoBroadcastJoinThreshold?

It is the size limit in bytes below which Spark automatically broadcasts a table — the default is 10 MB. Spark compares it against each side's estimated in-memory size; if one side is smaller, it plans a broadcast hash join. Set it to -1 to disable auto-broadcasting entirely. Remember the estimate is the in-memory size, which is often much larger than the compressed size on disk.

When should I use a shuffle hash join in Spark?

Use it when one side is too big to broadcast but still small enough to build an in-memory hash table per partition, and the keys are expensive to sort. It skips the sort step sort-merge pays, so it can win in that middle-ground case. Because it risks OOM when the build side is large, Spark disables it by default — enable it with a /*+ SHUFFLE_HASH(t) */ hint.

Does Adaptive Query Execution change the join strategy at runtime?

Yes. With AQE enabled (default since Spark 3.2), Spark can convert a planned sort-merge join into a broadcast join at runtime when a shuffle stage reveals one side is smaller than spark.sql.adaptive.autoBroadcastJoinThreshold. AQE also splits skewed partitions to fix skewed joins and coalesces small shuffle partitions — all using real statistics instead of compile-time estimates.

Conclusion

Spark join strategies come down to one question: can you avoid moving the big table? If one side is small, broadcast it and skip the shuffle — the single biggest win available in Spark tuning. If both sides are large, sort-merge is the dependable default, and Adaptive Query Execution will quietly upgrade it to a broadcast or fix skew when the real numbers come in. Shuffle hash join stays a rare, deliberate choice.

Master three habits and joins stop being the scary part of your pipeline: read the physical plan with .explain(), know your table sizes in memory (not on disk), and reach for a broadcast() hint the moment a small dimension is being shuffled. Do that and the same query that once ran for an hour finishes in minutes.

Want to go deeper on Spark and the modern data stack? Explore the free, hands-on Spark with Scala course and the rest of the SolutionGigs blog — every function and tuning technique with runnable examples. Building a data pipeline and want expert help? SolutionGigs connects you with vetted data engineers — it's free to post your project.

Mohammed Yaseen

Founder, SolutionGigs

Mohammed builds and tunes large-scale Spark pipelines on Kafka, EMR, and the lakehouse — and has spent more late nights reading physical plans than he'd like to admit. LinkedIn →