Why is my broadcast join running out of memory?

Summary

• A broadcast hash join holds the entire build side in memory: open-source Spark collects it to the driver, builds a hashed relation there, and then every executor keeps its own copy. An oversized build side can exhaust the driver, the executors, or both.
• One trigger is a bad size estimate: Spark compares spark.sql.autoBroadcastJoinThreshold (default 10 MB) with estimated statistics, so missing or stale statistics can let a large table through.
• A BROADCAST hint (aliases BROADCASTJOIN and MAPJOIN) broadcasts its side regardless of the threshold, and AQE can convert a sort-merge join to a broadcast join at runtime using spark.sql.adaptive.autoBroadcastJoinThreshold.
• The quickest fix is to stop the broadcast: set the thresholds to -1 or use a MERGE hint. Then fix the cause with ANALYZE TABLE ... COMPUTE STATISTICS, a narrower build side, or more driver memory.

Why is my broadcast join running out of memory?

Because the side being broadcast doesn't fit where Spark has to hold it. In open-source Apache Spark, a broadcast hash join collects the whole build side to the driver, builds a hash table from it there, and then sends a copy to every executor. If the real size of that side exceeds the driver heap or what each executor can spare, the job fails with an out-of-memory error. The causes to check: Spark underestimated the table, a hint forced the broadcast, or adaptive query execution (AQE) converted the join at runtime. Stop the broadcast first, then fix the cause.

Where the memory goes

The physical operator is BroadcastExchangeExec. In Spark 4.2.0 it calls executeCollectIterator() on the build side, checks the row limit, transforms the rows into a hashed relation on the driver, checks the byte limit, and then broadcasts the result. An OutOfMemoryError during that sequence surfaces as Spark's own error:

Not enough memory to build and broadcast the table to all worker nodes. As a workaround, you can either disable broadcast by setting spark.sql.autoBroadcastJoinThreshold to -1 or increase the spark driver memory by setting spark.driver.memory to a higher value ...

The same operator rejects a build side over its row or byte limit with "Cannot broadcast the table over ... rows" or "Cannot broadcast the table that is larger than ...". Once broadcast, the data behaves like a broadcast variable, a read-only value "cached on each machine", so every executor pays for its own copy.

Causes and fixes, by symptom

Symptom	Likely cause	Fix
Plan shows BroadcastHashJoin for a table you know is large	Missing or stale statistics put the estimate under spark.sql.autoBroadcastJoinThreshold	Run ANALYZE TABLE t COMPUTE STATISTICS; check estimates with EXPLAIN COST
A table over the threshold is broadcast anyway	A BROADCAST, BROADCASTJOIN or MAPJOIN hint, or .hint("broadcast"), which ignores the threshold	Remove the hint, or use MERGE to request a shuffle sort-merge join
Initial plan has SortMergeJoin, final plan has BroadcastHashJoin	AQE switched strategies using spark.sql.adaptive.autoBroadcastJoinThreshold, which defaults to the static threshold	Lower that setting, or set it to -1
BroadcastNestedLoopJoin in the plan	The join has no equi-join key, so the broadcast becomes a nested loop join	Add an equality condition, or drop the broadcast hint
The error names the driver	The build side is collected and hashed on the driver	Shrink the build side; raise spark.driver.memory (use --driver-memory in client mode)

Diagnose, then disable

Read the plan before touching memory settings. In the Spark UI's SQL tab, the BroadcastExchange node reports "data size" (the size of the broadcast data) and "time to collect" (the time spent collecting it).

joined = orders.join(customers, "customer_id")
joined.explain(mode="formatted")   # look for BroadcastExchange and BroadcastHashJoin

# Stop planned and AQE-converted broadcasts for this session
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", -1)

ANALYZE TABLE customers COMPUTE STATISTICS;
EXPLAIN COST SELECT * FROM orders o JOIN customers c ON o.customer_id = c.customer_id;
SELECT /*+ MERGE(c) */ * FROM orders o JOIN customers c ON o.customer_id = c.customer_id;

Then shrink what you broadcast. A SELECT * carries every column of the dimension through the exchange, so project only the columns you need and filter the dimension before the join. Keep -1 scoped to the session or query that needs it: setting it cluster-wide also turns off broadcasts for the small lookup tables they suit.

How Databricks approaches broadcast joins

Databricks Runtime 11.3 LTS and above use executor-side broadcast joins, which shift the memory pressure of building the broadcast from the driver to the executors. With that change, explicit broadcast hints on large tables can fail with "There is not enough memory to build the hash map", and the Databricks knowledge base recommends letting adaptive query execution choose the join instead of forcing one with hints. On Databricks, AQE's runtime switch to a broadcast join is governed by spark.databricks.adaptive.autoBroadcastJoinThreshold, whose documented default is 30MB; the Databricks AQE reference lists it with the other runtime settings.

FAQs

Does raising spark.sql.autoBroadcastJoinThreshold fix the error?

No. A higher threshold lets Spark broadcast larger tables, which increases the memory each broadcast needs. Lower it, or set it to -1 to disable automatic broadcasts.

Why does the driver run out of memory when I never call collect()?

In open-source Spark, the broadcast exchange collects the build side to the driver and builds the hashed relation there. That's why Spark's own error message suggests raising spark.driver.memory.

Is the 10 MB threshold measured on disk or in memory?

Spark compares the threshold with its estimate of the join side's size. Those estimates come from data source, catalog or runtime statistics, and missing or inaccurate statistics can put the estimate far from the real size of the hashed relation.

Can AQE cause a broadcast I didn't plan?

Yes. When a join side's runtime statistics fall under the adaptive threshold, AQE swaps the planned sort-merge join for a broadcast hash join. Set spark.sql.adaptive.autoBroadcastJoinThreshold to -1 to stop those conversions.