Apache Spark joins in depth: the five strategies and how one gets picked

• Types and strategies are different things
• Architecture: what every join is made of
• The five strategies
  • Broadcast hash join
  • Shuffle hash join
  • Sort-merge join
  • Broadcast nested loop join
  • Cartesian product
• How does Spark actually choose?
  • The size rules, exactly
  • Which build side each join type allows
  • Seeing it happen
• What adaptive execution changes
• Reading the plan
• Which strategy for which scenario
• Production tips
• Frequently asked questions
• Conclusion
• References
• Trademarks

TL;DR

• A join type is semantics (inner, outer, semi, anti). A join strategy is the algorithm (broadcast hash, shuffle hash, sort-merge, broadcast nested loop, cartesian). Mixing the two words up is why join tuning feels arbitrary.
• Spark’s choice is a documented, ordered procedure in JoinSelection, not a cost model. Hints first, then broadcast, then shuffle hash, then sort-merge, then cartesian, then nested loop “as the final solution”.
• The join type silently restricts the strategy. A full outer join can never be broadcast, and for a right outer join only the left side may be. Both are confirmed below by running them.
• Shuffle hash join needs three conditions at once, one of which is off by default, which is why you almost never see it without a hint.
• The 10 MB threshold is not the only size rule. Shuffle hash join has its own ceiling of autoBroadcastJoinThreshold multiplied by spark.sql.shuffle.partitions, which is 2 GB on the defaults.

Join tuning has a reputation for being folklore. Broadcast the small side, raise the threshold, add a hint, and if none of that works, salt the key. It works often enough to feel like knowledge and fails often enough to feel like luck.

The reason is that most of what circulates describes the five strategies without describing the procedure that picks between them. That procedure is not mysterious: it is written down, in order, in a comment above JoinSelection in Spark’s own source, and once you have read it the behaviour stops being surprising. A full outer join that refuses to broadcast is not a bug, and no amount of raising the threshold will change it.

This post has three parts. What each strategy actually does, the exact order Spark tries them in, and which scenario should get which. Every rule below was read from the v4.2.0 tag, and the behavioural claims were confirmed by running the joins and reading the chosen strategy out of the physical plan.

Types and strategies are different things

This is the distinction to get right first, because the two are chosen independently and the first one constrains the second.

A strategy change can make a query ten times faster or fail outright, and it cannot change the answer. A type change always changes the answer. So when someone says “we changed the join and it got faster”, the useful question is which of the two they changed.

The types, briefly, since the rest of the post assumes them:

trips.join(cities, "city_id")                      # inner, the default
trips.join(cities, "city_id", "left")              # left outer
trips.join(cities, "city_id", "full")              # full outer
trips.join(cities, "city_id", "left_semi")         # left rows that match
trips.join(cities, "city_id", "left_anti")         # left rows that do not
trips.crossJoin(cities)                            # every pair

Semi and anti deserve a word because they are under-used. Both filter the left side by whether a match exists and return no columns from the right, so a left row matching a hundred right rows is emitted once. An inner join in the same position emits it a hundred times and needs a distinct afterwards, which is a second shuffle.

Architecture: what every join is made of

All five strategies are assembled from the same three parts, and naming them makes the differences between the strategies easy to hold.

flowchart TB
  subgraph P["every join has these"]
    A["<b>build side</b><br/>materialised into a lookup<br/>structure, or sorted"]
    B["<b>stream side</b><br/>read once and probed<br/>against the build side"]
    C["<b>co-location step</b><br/>how matching keys are made<br/>to meet on one executor"]
  end
  A --> J["the join operator"]
  B --> J
  C --> J

The build side is the one Spark materialises: a hash table for the hash joins, a sorted run for sort-merge. It is the side that costs memory, which is why “which side is the build side” is the question behind most join failures.

The stream side is read once and probed. It costs nothing to hold, which is why you want the large table here.

The co-location step is where the strategies genuinely differ, and it is the expensive part:

Read down that column and the whole subject compresses to one trade: either you copy the small side to where the data already is, or you move both sides to a common place. Broadcasting does the first, the shuffle-based strategies do the second, and bucketing is how you pay for the second once at write time instead of on every read.

The five strategies

Spark’s source documents all five with their limits. The table below is that comment, condensed:

An equi-join is one whose condition is equality on keys, a.id = b.id. Anything else, a range or an inequality, is a non-equi join, and only the bottom two rows can execute it at all. That single fact explains most surprises in the plan.

Broadcast hash join

Spark collects the small side to the driver, sends a copy to every executor, and each executor builds a hash table from it. Each partition of the large side then probes that table locally. The large side is never shuffled.

flowchart TB
  D["driver collects the small side"] --> E1["executor 1<br/>hash table"]
  D --> E2["executor 2<br/>hash table"]
  D --> E3["executor 3<br/>hash table"]
  L1["large partition"] --> E1
  L2["large partition"] --> E2
  L3["large partition"] --> E3

Cost: one copy of the small side per executor, held in memory for the duration, plus the driver having to assemble it first.

That driver step is the part people forget. A broadcast that is too large fails on the driver, not the executors, and the error often arrives as a spark.sql.broadcastTimeout after the default 300 seconds rather than as an obvious out-of-memory.

Shuffle hash join

Both sides are shuffled on the join key. Then, for each partition, Spark builds a hash table from the smaller side and probes it with the larger. No sorting.

flowchart TB
  A["side A"] --> SH[("shuffle by key")]
  B["side B"] --> SH
  SH --> P1["partition 0:<br/>hash small, probe large"]
  SH --> P2["partition 1:<br/>hash small, probe large"]

It sits in a narrow band: one side too big to broadcast, but small enough that a per-partition hash table fits. It skips the sort that sort-merge pays, and in exchange it can run out of memory where sort-merge would spill.

Sort-merge join

Both sides are shuffled on the key, each partition is sorted, and the two sorted streams are walked together with a cursor on each.

flowchart TB
  A["side A"] --> S[("shuffle by key")]
  B["side B"] --> S
  S --> SA["sort partition"]
  SA --> M["merge: advance the<br/>smaller cursor, emit matches"]

left  (sorted):  a a b c d ...
right (sorted):  a b b d e ...
                 ^ ^

Its virtue is bounded memory. Only the merge buffers are held, not a whole hash table, and the sort can spill, so a partition larger than memory still finishes. That is why it is the default for large-to-large joins, and why spark.sql.join.preferSortMergeJoin defaults to true.

The cost is the sort, and the requirement is that the keys are sortable. It is the only strategy with that constraint.

Broadcast nested loop join

One side is broadcast, and then every row of one side is compared against every row of the other, evaluating an arbitrary condition. It is the fallback that makes non-equi joins possible at all.

The source notes it is optimised for three specific shapes, and slow otherwise:

1. Broadcasting the left side in a right outer join.
2. Broadcasting the right side in a left outer, left semi, left anti or existence join.
3. Broadcasting either side in an inner-like join.

Outside those, “we need to scan the data multiple times, which can be rather slow”. Seeing BroadcastNestedLoopJoin on two large inputs means the query is doing something close to a cross product.

Cartesian product

Also called shuffle-and-replicate nested loop. Every partition of one side is paired with every partition of the other, producing the full cross product. It supports inner-like joins only, and it is what you get from a deliberate crossJoin once broadcasting is off the table.

How does Spark actually choose?

Here is the part that turns folklore into a procedure. JoinSelection follows a fixed order, and the first applicable rule wins.

For an equi-join, hints are consulted first, in this order:

1. BROADCAST: pick broadcast hash join if the join type supports it.
2. MERGE: pick sort-merge join if the keys are sortable.
3. SHUFFLE_HASH: pick shuffle hash join if the join type supports it.
4. SHUFFLE_REPLICATE_NL: pick cartesian product if the join type is inner-like.

With no hints, for an equi-join:

1. Broadcast hash join, if one side is small enough to broadcast and the join type allows that side to be the build side.
2. Shuffle hash join, if one side can build a local hash map, is much smaller than the other, and spark.sql.join.preferSortMergeJoin is false.
3. Sort-merge join, if the join keys are sortable.
4. Cartesian product, if the join type is inner-like.
5. Broadcast nested loop join as the last resort. The source is blunt about it: “It may OOM but we don’t have other choice.”

For a non-equi join the list is much shorter, which is why these are so often slow:

1. Broadcast nested loop join, if one side is small enough to broadcast.
2. Cartesian product, if the join type is inner-like.
3. Broadcast nested loop join anyway.

flowchart TB
  H{"a join hint?"} -->|"yes"| HH["honour it if the type<br/>and keys allow"]
  H -->|"no"| EQ{"equi-join?"}
  EQ -->|"no"| NE["broadcast nested loop if a side is small,<br/>else cartesian if inner-like,<br/>else nested loop anyway"]
  EQ -->|"yes"| B{"a side small enough,<br/>and allowed as build side?"}
  B -->|"yes"| BHJ["BroadcastHashJoin"]
  B -->|"no"| SH{"local hash map fits,<br/>much smaller,<br/>preferSortMergeJoin false?"}
  SH -->|"yes"| SHJ["ShuffledHashJoin"]
  SH -->|"no"| SM{"keys sortable?"}
  SM -->|"yes"| SMJ["SortMergeJoin"]
  SM -->|"no"| C["cartesian if inner-like,<br/>else nested loop"]

The size rules, exactly

Three predicates decide “small enough”, and only the first is widely known.

Broadcast. canBroadcastBySize is sizeInBytes >= 0 && sizeInBytes <= autoBroadcastJoinThreshold, where the threshold is spark.sql.autoBroadcastJoinThreshold, default 10485760, which is 10 MB.

There is a detail worth having: when the statistics are runtime statistics, meaning adaptive execution has measured a completed stage, Spark uses spark.sql.adaptive.autoBroadcastJoinThreshold instead, falling back to the static one when that is unset. So plan-time and runtime broadcasting can be governed by two different numbers.

Shuffle hash, ceiling. canBuildLocalHashMapBySize is sizeInBytes < autoBroadcastJoinThreshold * numShufflePartitions. On the defaults that is 10 MB multiplied by 200:

10,485,760 x 200 = 2,097,152,000 bytes = ~2 GB

So the shuffle-hash ceiling is roughly 2 GB by default, two hundred times the broadcast threshold, and it moves when you change the shuffle partition count.

Shuffle hash, ratio. muchSmaller is a.size * spark.sql.shuffledHashJoinFactor <= b.size, and that factor defaults to 3. The source explains why: “The cost to build hash map is higher than sorting, we should only build hash map on a table that is much smaller than other one.”

So shuffle hash join requires three things simultaneously: under the 2 GB ceiling, at least three times smaller than the other side, and preferSortMergeJoin flipped to false. That conjunction, with the last condition off by default, is why it is rare in practice.

Which build side each join type allows

This is the rule that makes broadcast joins look unpredictable, and it is pure lookup, not heuristics.

The logic follows from correctness. A left outer join must emit every left row, so the left side has to be streamed and the right one built. A right outer is the mirror image. A full outer must emit unmatched rows from both sides, so neither can be the build side, and no full outer join is ever a broadcast hash join regardless of how small either side is.

Shuffle hash join is less restricted, which is why the strategy table says it supports all join types: its build side may be the left for inner, left outer, full outer and right outer, and the right for those plus left semi, left anti and existence joins.

Seeing it happen

The rules above are testable, so here they are run on Spark, reading the chosen strategy out of the executed plan:

def strategy(df):
    plan = df._jdf.queryExecution().executedPlan().toString()
    for name in ["BroadcastHashJoin", "ShuffledHashJoin", "SortMergeJoin",
                 "BroadcastNestedLoopJoin", "CartesianProduct"]:
        if name in plan:
            return name
    return "unknown"

The two rows in bold are the ones worth remembering, because both look like Spark ignoring an obvious optimisation:

• Full outer with a tiny side still sort-merges. Neither side may be the build side, so broadcasting is not available at any threshold.
• Right outer with the small side on the right still sort-merges. Only the left may be broadcast for a right outer join. Put the small table on the left, or use a left outer join with the sides swapped, and it broadcasts.

One more result is instructive. With broadcasting disabled and preferSortMergeJoin set to false, a 200,000-row side against a 3,000,000-row side still chose SortMergeJoin, because it was not three times smaller in bytes. Shuffle hash join really does need all three conditions.

What adaptive execution changes

Everything above is plan-time. Adaptive query execution re-plans after a shuffle stage completes, when the sizes are measurements rather than estimates, and spark.sql.adaptive.enabled defaults to true.

For joins it does two things:

1. Demotes a sort-merge join to a broadcast hash join when a side turns out small enough, using spark.sql.adaptive.autoBroadcastJoinThreshold.
2. Splits skewed partitions, with spark.sql.adaptive.skewJoin.enabled default true.

The skew thresholds are strict and worth knowing, because they explain “skew handling is on but nothing happened”. A partition is treated as skewed only when it is larger than spark.sql.adaptive.skewJoin.skewedPartitionFactor, default 5.0, times the median and larger than spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes, default 256MB. On a job whose partitions are all under 256 MB, it never fires no matter how uneven they are.

The practical consequence for reading plans: explain() shows the planned shape, not the final one. Before execution the tree is wrapped in AdaptiveSparkPlan with isFinalPlan=false. The SQL tab of the UI after the query finishes shows what actually ran.

Reading the plan

df.explain("formatted")

What to look for, in order of how much it tells you:

A useful habit: count the Exchange nodes. That is the number of shuffles, and if two consecutive operations key on the same column, one exchange can serve both.

Which strategy for which scenario

Two of those need expanding.

The range join. A condition like ON t.ts BETWEEN d.start AND d.end has no equality, so Spark can only nested-loop it. The standard fix is to manufacture a key: bucket both sides by a coarser value, join on that equality and the range, so the comparison happens within each bucket rather than across the whole product.

(events.withColumn("day", F.to_date("ts"))
   .join(windows.withColumn("day", F.to_date("start")),
         ["day"])                                    # an equality to partition on
   .filter((F.col("ts") >= F.col("start")) & (F.col("ts") <= F.col("end"))))

The full outer with a small side. If broadcasting matters more than the single scan, a full outer can be rewritten as a left outer union an anti join the other way, both of which can broadcast. It is more code and two passes, so only worth it when the sort-merge is genuinely the bottleneck.

Production tips

• Read the plan before tuning. The strategy Spark chose tells you which rule fired, and therefore which lever exists.
• Reduce before joining. Filtering and projecting a side is the highest-value change available, because it can move the join into the broadcast band and change the strategy entirely.
• Do not hint BROADCAST on something large. A hint overrides the size check, so you move the failure to the driver, which must collect it first.
• Treat a full outer join as unbroadcastable. Raising the threshold will never help; the join type is what forbids it.
• Check which side is small for outer joins. Broadcast eligibility is asymmetric, and swapping the sides is often the whole fix.
• Fix the estimate rather than forcing the plan where you can, with ANALYZE TABLE ... COMPUTE STATISTICS, so Catalyst chooses correctly on its own.
• Bucket the keys you join on repeatedly. It is the only way to remove the shuffle from a large-to-large join permanently.
• Trust the SQL tab over explain() for what actually ran, since AQE re-plans after the shuffle.

Frequently asked questions

Why is my small table not being broadcast? Four common reasons: the estimate is of the unfiltered table because the filter did not push down; the table is compressed on disk and larger in memory than you think; there are no statistics so Spark fell back to file size; or the join type forbids that side as the build side. Check the last one first, because no config change fixes it.

Does a bigger autoBroadcastJoinThreshold make joins faster? Sometimes, and it moves the failure mode from slow to fragile. Every executor holds a full copy and the driver assembles it first, so a 500 MB broadcast across 50 executors is 25 GB of cluster memory plus a driver that must hold it. Raise it deliberately, with a number you can justify.

Why do I almost never see ShuffledHashJoin? Because it needs three conditions at once and one of them, spark.sql.join.preferSortMergeJoin being false, is not the default. Spark prefers sort-merge because it cannot run out of memory, while a hash join can.

Can a join strategy change my results? No. Strategies are algorithms for the same semantics. If results changed, the join type or the condition changed, or there are nulls in the key behaving as SQL says they should.

Why is my join a CartesianProduct when I wrote a condition? Because Catalyst could not use the condition as an equality: it is an inequality, it is wrapped in an expression it cannot match on, or it sits in a where after a cross join in a way that could not be pushed into the join.

What about joining on a collated string column? In Spark 4 this matters. Hash joins require keys that are binary-stable, and a case- or accent-insensitive collation is not, because two rows can be equal without having equal bytes. Joining on such a column disables hash joins and leaves sort-merge.

Conclusion

The thing worth taking away is that join selection is a lookup, not a judgement. Spark asks, in a fixed order: is there a hint, is this an equi-join, is one side small enough, does the join type permit that side as the build side, are the keys sortable. The first rule that fits wins. Nothing about it is adaptive except the parts AQE re-plans after a shuffle.

That reframes tuning. You are not persuading an optimiser, you are changing the inputs to a decision procedure. Filtering a dimension until it fits under 10 MB changes the answer to “is one side small enough”. Swapping the sides of an outer join changes the answer to “may that side be the build side”. Bucketing changes whether a shuffle is needed at all. Each of those is a specific rule with a specific lever.

And two answers are simply no. A full outer join will not broadcast, and a non-equi join will not hash. When you hit those, the fix is to change the query shape, not the configuration, which is exactly the kind of thing worth knowing before spending an afternoon on thresholds.

References

• SparkStrategies.scala at v4.2.0, whose JoinSelection comment documents the five strategies and the order they are tried in
• joins.scala at v4.2.0 for the size predicates and the build-side rules per join type
• SQL performance tuning for join hints and adaptive query execution
• Apache Spark architecture for the shuffle and stage machinery these strategies sit on
• 200 Apache Spark interview questions which covers joins among much else

Trademarks

Apache Spark, Apache Hive, Apache Parquet and Apache are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.