Broadcast join versus sort-merge join: how does Spark choose?

Summary

• With no usable hint, Spark picks a broadcast hash join when one side is small enough to broadcast and the join type allows it, then a shuffle hash join under extra conditions, then a sort-merge join if the join keys are sortable.
• The size test uses spark.sql.autoBroadcastJoinThreshold, 10 MB (10485760 bytes) by default. Setting it to -1 disables automatic broadcasting.
• A sort-merge join shuffles both relations by the join keys and sorts them. A broadcast hash join does not shuffle the streamed relation.
• Hints are suggestions. Spark prioritizes BROADCAST over MERGE over SHUFFLE_HASH over SHUFFLE_REPLICATE_NL and does not guarantee the hinted strategy.
• Adaptive Query Execution (AQE), enabled by default since Spark 3.2.0, can convert a sort-merge join to a broadcast hash join at runtime.

Broadcast join versus sort-merge join: how does Spark choose?

Spark chooses a join strategy for each join from the join hints, the join type, whether the join keys are sortable, and the sizes of the relations. For an equi-join with no usable hint, the planner applies a fixed list of rules in order. The size test for broadcasting is spark.sql.autoBroadcastJoinThreshold, which defaults to 10485760 bytes (10 MB).

How do the two strategies differ?

Aspect	Broadcast hash join	Sort-merge join
How it works	The build side is placed in a Spark broadcast variable, and the streamed relation is not shuffled	Both relations are shuffled using the join keys, and each side is sorted by its join keys
Join keys	Equi-joins only; keys need not be sortable	Equi-joins only; keys must be sortable
Join types	All join types except full outer joins	All join types
Trade-off	Broadcasting is network-intensive and can cause out-of-memory errors when the build side is big	Shuffle and sort work on both relations
Hint	BROADCAST (aliases BROADCASTJOIN, MAPJOIN)	MERGE (aliases SHUFFLE_MERGE, MERGEJOIN)

How does Spark choose when there is no hint?

For an equi-join with no hint, or with hints that do not apply, the planner follows these rules one by one:

1. Pick a broadcast hash join if one side is small enough to broadcast and the join type is supported. If both sides are small, broadcast the smaller side by statistics.
2. Pick a shuffle hash join if one side is small enough to build a local hash map, is much smaller than the other side, and spark.sql.join.preferSortMergeJoin is false.
3. Pick a sort-merge join if the join keys are sortable.
4. Pick a cartesian product if the join type is inner-like.
5. Pick a broadcast nested loop join as the final option. The source comment notes that it may run out of memory but there is no other choice.

Hints are checked first. Spark prioritizes BROADCAST over MERGE over SHUFFLE_HASH over SHUFFLE_REPLICATE_NL when different hints sit on the two sides of a join.

How do I control the choice?

-- BROADCAST, BROADCASTJOIN and MAPJOIN are accepted for the broadcast hint
SELECT /*+ BROADCAST(r) */ * FROM src s JOIN records r ON s.key = r.key;

-- Suggest a sort-merge join
SELECT /*+ MERGE(t1) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.key;

spark.table("src").join(spark.table("records").hint("broadcast"), "key").show()

# Disable automatic broadcasting; a BROADCAST hint still applies
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

A hint only suggests a strategy. A strategy may not support every join type, so Spark is not guaranteed to use the one you hint. The broadcast hint broadcasts the hinted side regardless of autoBroadcastJoinThreshold. If a table has an alias, the hint must name the alias, as BROADCAST(r) does above.

What does AQE change at runtime?

Adaptive Query Execution uses runtime statistics to choose the execution plan. It has been enabled by default since Apache Spark 3.2.0, and spark.sql.adaptive.enabled is the umbrella setting. When the runtime statistics of any join side fall below the adaptive broadcast hash join threshold, AQE converts a sort-merge join to a broadcast hash join. The docs say this is not as efficient as planning a broadcast hash join in the first place, but it is better than continuing the sort-merge join, because it avoids sorting both join sides and can read shuffle files locally to save network traffic.

Property	Default	Meaning
spark.sql.autoBroadcastJoinThreshold	10485760 (10 MB)	Maximum size in bytes of a table broadcast in a join; -1 disables broadcasting
spark.sql.broadcastTimeout	300	Timeout in seconds for the broadcast wait time in broadcast joins
spark.sql.adaptive.autoBroadcastJoinThreshold	none; same as spark.sql.autoBroadcastJoinThreshold	Broadcast threshold used only in the adaptive framework (since 3.2.0)
spark.sql.adaptive.localShuffleReader.enabled	true	Reads shuffle data locally when shuffle partitioning is not needed, for example after converting a sort-merge join to a broadcast hash join

How do I see which join Spark used?

Call df.explain() or run EXPLAIN in SQL to print the physical plan, then read the join operator. Physical nodes drop the Exec suffix from their class names, so the operators are BroadcastHashJoin (class BroadcastHashJoinExec) and SortMergeJoin (class SortMergeJoinExec). The SQL tab of the Spark UI lists each query as a graph of operators with their metrics. Because AQE can change a join at runtime, compare the initial plan with the executed query.

How does Databricks approach join strategy selection?

Databricks documents adaptive query execution as enabled by default, controlled by spark.databricks.optimizer.adaptive.enabled (default true). Its runtime switch to a broadcast join has its own property, spark.databricks.adaptive.autoBroadcastJoinThreshold, which the docs list with a default of 30MB.

FAQs

When should I add a BROADCAST hint?

Add it when you know one side is small but the statistics do not show it. Spark prioritizes the hinted side for broadcast even if the size suggested by the statistics is above the threshold.

Why did my sort-merge join not become a broadcast join?

Check that AQE is enabled, that a side's runtime size actually fell below the adaptive threshold, that the threshold is not -1, and that the join type supports broadcasting. Broadcast hash joins do not support full outer joins.