How do I broadcast a small table to avoid a shuffle join?

Summary

• Wrap the small DataFrame in broadcast(), or add a /*+ BROADCAST(t) */ hint in SQL, and Spark plans a broadcast join: the small side goes to every executor and the large side is joined where it sits, without a shuffle.
• Without any hint, a side whose estimated size falls below spark.sql.autoBroadcastJoinThreshold (10 MB by default) is broadcast automatically. A hint forces the broadcast regardless of that threshold.
• The join type limits which side a broadcast hash join can build on: the right side for left outer, left semi and left anti joins, the left side for right outer joins, and neither side of a full outer join.
• Confirm it with explain(): look for BroadcastHashJoin and a BroadcastExchange on the small side, with no Exchange on the large side.
• If Spark keeps choosing a sort-merge join, check the statistics it plans with. Adaptive Query Execution can still switch to a broadcast at runtime once a shuffle shows one side is small.

How do I broadcast a small table to avoid a shuffle join?

Mark the small side for broadcast. In PySpark, pyspark.sql.functions.broadcast() "marks a DataFrame as small enough for use in broadcast joins"; in SQL, the BROADCAST hint does the same. Spark then sends a full copy of the small table to every executor and joins each partition of the large table locally, so the large side never passes through a shuffle. Use it when one side really is small, because every executor receives the whole relation.

Step 1: Request the broadcast

from pyspark.sql import functions as sf

orders = spark.table("sales.orders")    # large
regions = spark.table("sales.regions")  # small

joined = orders.join(sf.broadcast(regions), "region_id")
joined.explain()

SELECT /*+ BROADCAST(r) */ o.order_id, r.region_name
FROM sales.orders o
JOIN sales.regions r ON o.region_id = r.region_id;

df.hint("broadcast") is equivalent in the DataFrame API, and Scala has org.apache.spark.sql.functions.broadcast. In SQL, BROADCASTJOIN and MAPJOIN are aliases for BROADCAST. The Spark docs say the hinted side "will be broadcast regardless of autoBroadcastJoinThreshold". If both sides carry the hint, Spark broadcasts the one that's smaller by its statistics.
Shrink the small side before you broadcast it. A broadcast exchange collects the result of the plan beneath it and then broadcasts it, so a projection and filter applied first reduce what every executor has to hold.

Step 2: Check that the join type allows it

In a broadcast hash join, the broadcast table is the build side and the other table is streamed through without a shuffle. Spark only allows certain build sides:

Join type	Build side a broadcast hash join allows	What it means for you
Inner	Either side	Broadcast whichever table is small
Left outer, left semi, left anti	Right side only	Put the small table on the right
Right outer	Left side only	Put the small table on the left
Full outer	Neither side	With an equi-join key, expect a shuffle-based join such as sort-merge

When the hinted side can't be the build side, Spark won't use the hinted strategy; the docs state that Spark "is not guaranteed to use the join strategy suggested by the hint." A BROADCAST hint on a join without an equi-join key produces a broadcast nested loop join instead of a broadcast hash join.

Step 3: Verify the plan

Run explain() before the action. A successful broadcast shows BroadcastHashJoin with a BroadcastExchange under the small side and no Exchange above the large side's scan. A SortMergeJoin with an Exchange on both inputs means both sides are still shuffled. After the query runs, the SQL tab of the Spark UI shows the BroadcastExchange node with its data size and time to collect.

Step 4: Know when Spark decides for you

Without a hint, Spark broadcasts a side whose estimated size is under spark.sql.autoBroadcastJoinThreshold. Its default is 10485760 bytes (10 MB), and -1 disables automatic broadcasting. The estimate comes from statistics, and the Spark docs warn that missing or inaccurate statistics hinder plan selection. Refresh catalog statistics with ANALYZE TABLE and inspect the estimates with DataFrame.explain(mode="cost").
Adaptive Query Execution (AQE), on by default since Spark 3.2.0, adds a second chance. After a shuffle, AQE converts a sort-merge join to a broadcast hash join when a side's runtime size is under spark.sql.adaptive.autoBroadcastJoinThreshold, which defaults to the same value as the static threshold. The docs note this isn't as efficient as planning the broadcast up front, though it avoids sorting both sides and reads shuffle files locally when spark.sql.adaptive.localShuffleReader.enabled is true. The broadcast wait is bounded by spark.sql.broadcastTimeout, 300 seconds by default.

How Databricks approaches broadcast joins

Databricks SQL and Databricks Runtime accept the same join hints, with BROADCASTJOIN and MAPJOIN as aliases. AQE is enabled by default on Databricks, and its runtime switch to a broadcast hash join uses spark.databricks.adaptive.autoBroadcastJoinThreshold, documented with a default of 30MB. On serverless compute, which doesn't expose the Spark UI, use the query profile to see which join operator ran and how much time and memory it took.

FAQs

Does a broadcast hint override the size threshold?

Yes. The hinted side is broadcast regardless of spark.sql.autoBroadcastJoinThreshold, as long as the join type allows that side to be the build side.

How do I stop Spark from broadcasting a table?

Set spark.sql.autoBroadcastJoinThreshold to -1 to turn off automatic broadcasts, or use the MERGE hint to request a sort-merge join for a specific query.

What happens if both sides of the join have a broadcast hint?

Spark broadcasts the side that's smaller according to its statistics. With different hints on each side, BROADCAST takes priority over MERGE, SHUFFLE_HASH and SHUFFLE_REPLICATE_NL.