What is a broadcast join?

Summary

• A broadcast join copies the smaller side of a join, the build side, to every worker node, so Apache Spark can join it with the larger side without shuffling the larger side across the network.
• Spark broadcasts a table automatically when its estimated size is no larger than spark.sql.autoBroadcastJoinThreshold, which defaults to 10 MB. Setting it to -1 disables automatic broadcasting.
• The BROADCAST join hint (aliases BROADCASTJOIN and MAPJOIN), DataFrame.hint("broadcast") and functions.broadcast() request a broadcast regardless of that threshold, though Spark may pick another strategy if the join type doesn't support it.
• Adaptive Query Execution, on by default since Spark 3.2.0, can convert a sort-merge join to a broadcast hash join at runtime when a join side turns out to be small.

What is a broadcast join?

A broadcast join is a join strategy in which Spark sends a full copy of one relation, the build side, to every worker node. The other, larger relation is then joined where its partitions already sit, so it doesn't need a shuffle. Spark's tuning guide defines the size limit as "the maximum size in bytes for a table that will be broadcast to all worker nodes when performing a join."
Spark has two broadcast strategies: a broadcast hash join when the join has an equi-join key, and a broadcast nested loop join when it doesn't.
A sort-merge join instead shuffles and sorts both inputs by the join key. The example plan in Spark's tuning guide shows an Exchange (a data shuffle) and a Sort under each side of a SortMergeJoin.

When does Spark broadcast a table automatically?

Spark checks each join side against spark.sql.autoBroadcastJoinThreshold:

• Default: 10485760 bytes (10 MB). A value of -1 disables automatic broadcasting.
• The size is an estimate. Spark gets statistics from the data source (such as Parquet file metadata), from the catalog (collected by ANALYZE TABLE), or at runtime through Adaptive Query Execution (AQE). The tuning guide warns that missing or inaccurate statistics "will hinder Spark's ability to select an optimal plan."
• The join type limits the build side. Broadcast isn't supported for certain join types. For example, the left relation of a LEFT OUTER JOIN can't be broadcast.
• Timeout: spark.sql.broadcastTimeout (default 300) is the broadcast wait timeout, in seconds.

To refresh catalog statistics and see what Spark estimates:

ANALYZE TABLE records COMPUTE STATISTICS FOR ALL COLUMNS;
EXPLAIN COST SELECT * FROM src s JOIN records r ON s.key = r.key;

How do you ask Spark for a broadcast join?

Add a join hint in SQL, call hint("broadcast") on a DataFrame, or wrap the smaller DataFrame in broadcast():

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

from pyspark.sql import functions as sf

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

df = spark.createDataFrame([1, 2, 3, 3, 4], "int")
df_small = spark.range(3)
df.join(sf.broadcast(df_small), df.value == df_small.id).show()

The hinted side "will be broadcast regardless of autoBroadcastJoinThreshold." If both sides carry a broadcast hint, Spark broadcasts the smaller one based on statistics. A hint isn't a guarantee, because a given strategy may not support every join type.

Which join strategy hints does Spark support?

Hint	Aliases	What it asks Spark to use
BROADCAST	BROADCASTJOIN, MAPJOIN	Broadcast join. The hinted side is broadcast regardless of autoBroadcastJoinThreshold
MERGE	SHUFFLE_MERGE, MERGEJOIN	Shuffle sort merge join
SHUFFLE_HASH	None	Shuffle hash join. If both sides are hinted, the smaller side (based on stats) is the build side
SHUFFLE_REPLICATE_NL	None	Shuffle-and-replicate nested loop join

When the two sides of a join carry different hints, Spark prioritizes BROADCAST over MERGE over SHUFFLE_HASH over SHUFFLE_REPLICATE_NL, and it logs a warning that the overridden hint won't take effect. Before Spark 3.0, BROADCAST was the only join hint.

How does Adaptive Query Execution change the plan?

AQE re-optimizes a query plan during execution from runtime statistics, and it has been on by default since Spark 3.2.0 (spark.sql.adaptive.enabled). It converts a sort-merge join to a broadcast hash join when the runtime statistics of any join side are smaller than spark.sql.adaptive.autoBroadcastJoinThreshold. That setting has no value of its own by default, so it uses spark.sql.autoBroadcastJoinThreshold.
The tuning guide notes that a runtime conversion isn't as efficient as planning a broadcast hash join in the first place. It still avoids sorting both join sides, and it reads shuffle files locally when spark.sql.adaptive.localShuffleReader.enabled is true (the default). If you know a side is small, a hint lets Spark plan the broadcast up front.

How do you check whether a broadcast happened?

Call explain(). In the PySpark DataFrame.hint example, the plain join plans a SortMergeJoin and the same join with hint("broadcast") plans a BroadcastHashJoin. df.explain(mode="formatted") splits the output into a physical plan outline and node details.
Because AQE can change the plan at runtime, also check the Spark UI's SQL tab while or after the query runs. Its Details link shows the full plan, and runtime statistics appear as Statistics(..., isRuntime=true).

How does Databricks handle broadcast joins?

On Databricks, AQE is on by default for non-streaming queries that contain at least one exchange or subquery. The runtime switch to a broadcast hash join uses spark.databricks.adaptive.autoBroadcastJoinThreshold, which defaults to 30MB. Databricks' AQE guide says a broadcast hint can still be a good choice if you know your query well, because AQE might not switch to a broadcast join until after it has shuffled both sides. It also skips the conversion when too few partitions are non-empty, per spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin.

What else do people ask?

Can I broadcast a large table?

A hint bypasses the size threshold, but Spark still copies the whole relation to every worker node, and the broadcast must finish within spark.sql.broadcastTimeout. Save explicit hints for sides you know are small.

Does a broadcast join need an equality condition?

A broadcast hash join does. Without an equi-join key, a BROADCAST hint gives a broadcast nested loop join instead.