How do I force or tune a broadcast join?

Summary

• Spark broadcasts a join side automatically when its estimated size is within spark.sql.autoBroadcastJoinThreshold, which defaults to 10485760 bytes (10 MB). Set it to -1 to turn automatic broadcasting off.
• To force a broadcast regardless of the threshold, use the BROADCAST hint in SQL (aliases BROADCASTJOIN and MAPJOIN) or .hint("broadcast") on a DataFrame. The broadcast() function marks a DataFrame as small enough for broadcast joins.
• Spark doesn't guarantee a hinted strategy, because a strategy may not support every join type.
• Adaptive Query Execution (AQE), on by default since Spark 3.2.0, can turn a sort-merge join into a broadcast hash join at runtime.
• Confirm the result with df.explain(): look for BroadcastHashJoin and BroadcastExchange.

How do I force or tune a broadcast join?

In a broadcast hash join, Spark sends the small side to the executors and uses it as the build side, which removes the shuffle a sort-merge join needs. Spark plans one automatically when a side's estimated size is within spark.sql.autoBroadcastJoinThreshold (default 10485760 bytes, or 10 MB). To force one, mark the small side with the BROADCAST hint, .hint("broadcast") or the broadcast() function. To tune it, change the threshold, or set it to -1 to disable automatic broadcasting.

How does Spark decide to broadcast automatically?

The threshold is "the maximum size in bytes for a table that will be broadcast to all worker nodes when performing a join". Spark compares it with a size estimate built from statistics: from the data source (such as Parquet metadata), from the catalog (updated by ANALYZE TABLE), or computed at runtime by AQE. Missing or inaccurate statistics hinder plan selection, so check the estimates with EXPLAIN COST or df.explain(mode="cost"). ANALYZE TABLE ... COMPUTE STATISTICS collects the row count and size in bytes, and NOSCAN collects only the size.

How do I force a broadcast join?

from pyspark.sql.functions import broadcast

src, records = spark.table("src"), spark.table("records")
src.join(records.hint("broadcast"), "key").explain()  # hint on the small side
src.join(broadcast(records), "key").explain()         # marks records as small enough

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "52428800")  # bytes
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")        # automatic off

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

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

With a BROADCAST hint, Spark prioritizes a broadcast join with the hinted table as the build side, even if its estimated size is above the threshold. It uses a broadcast hash join when there is an equi-join key and a broadcast nested loop join when there isn't. When both sides carry different strategy hints, BROADCAST wins over MERGE, which wins over SHUFFLE_HASH, then SHUFFLE_REPLICATE_NL. If both sides are hinted BROADCAST, Spark broadcasts the smaller side based on statistics. There is no guarantee Spark follows a hint, because a strategy may not support every join type.

How do I tune the threshold and AQE?

Raising spark.sql.autoBroadcastJoinThreshold lets larger tables be broadcast to all worker nodes, so check the build side's size with EXPLAIN COST before you raise it. Setting it to -1 stops automatic broadcasting, but a hinted side is still broadcast "regardless of autoBroadcastJoinThreshold".
AQE, controlled by spark.sql.adaptive.enabled (default true), converts a sort-merge join to a broadcast hash join when runtime statistics of a join side fall below spark.sql.adaptive.autoBroadcastJoinThreshold. That setting defaults to the value of spark.sql.autoBroadcastJoinThreshold and accepts -1 to disable the conversion. With spark.sql.adaptive.localShuffleReader.enabled (default true), the converted join reads shuffle files locally. The Spark docs note that this conversion isn't as efficient as planning the broadcast hash join in the first place. spark.sql.broadcastTimeout (default 300) is the timeout in seconds for the broadcast wait time.

Symptom	Likely cause	Fix
The plan shows SortMergeJoin though one side is small	The size estimate is above the threshold, or statistics are missing	Run ANALYZE TABLE, check EXPLAIN COST, or add a BROADCAST hint
A hint has no effect	The join type can't broadcast that side (for example, the left side of a LEFT OUTER JOIN), or another hint overrides it	Hint the other side, or remove the conflicting hint
The broadcast fails on a timeout	The broadcast wait exceeded spark.sql.broadcastTimeout	Shrink the build side with filters or fewer columns, or raise the timeout
The broadcast only appears in the final adaptive plan	AQE converted the join at runtime	Hint the small side so the broadcast is planned up front

How do I check whether a join was broadcast?

df.explain() prints the physical plan. A broadcast join shows BroadcastHashJoin with a BroadcastExchange under it, while a shuffled join shows SortMergeJoin over Exchange hashpartitioning on both inputs. With AQE on, the root is an AdaptiveSparkPlan node, and isFinalPlan=false means execution hasn't finished. After the query runs, the SQL / DataFrame tab of the Spark UI reports "data size" and "time to collect" for each BroadcastExchange.

How does Databricks approach broadcast joins?

On Databricks, AQE is on by default and switches to a broadcast join at runtime based on spark.databricks.adaptive.autoBroadcastJoinThreshold (default 30MB). The AQE guide says a broadcast hint can still be a good choice when you know the query well, because AQE might not switch until after it shuffles both sides, and AQE respects query hints. Join hints apply to both Databricks SQL and Databricks Runtime. Databricks generally recommends against configuring most Spark properties, and serverless notebooks and jobs accept only a short list that doesn't include the broadcast thresholds, so use hints there. To keep size estimates fresh, predictive optimization runs ANALYZE on Unity Catalog managed tables and collects statistics when data is written.

FAQs

Can I broadcast both sides of a join?

No. If both sides carry a BROADCAST hint, Spark broadcasts the smaller one based on statistics.

Does a threshold of -1 block hinted broadcasts?

No. It only disables automatic broadcasting. A hinted side is still broadcast.

Are BROADCAST, BROADCASTJOIN and MAPJOIN different?

No. BROADCASTJOIN and MAPJOIN are aliases for BROADCAST.