Broadcast joins

How it works

1. Collect

   The small side is computed and collected to the driver

2. →

   Broadcast

   The driver sends it to every executor

3. →

   Build

   Each executor builds a hash table from it

4. →

   Probe

   Each task streams its large-side partition through the hash table

Compare a sort-merge join of two large tables: both sides shuffled by key, both sorted, then merged. Broadcasting replaces two shuffles and two sorts with one small transfer.

When Spark broadcasts

• Automatically, when its estimated size is below spark.sql.autoBroadcastJoinThreshold (10 MB by default; -1 disables it).
• With AQE, at runtime, when the measured size after a shuffle is below spark.sql.adaptive.autoBroadcastJoinThreshold, which defaults to the same value.
• When you ask: F.broadcast(df) or a SQL hint. Hints override the threshold.

from pyspark.sql import functions as F

orders.join(F.broadcast(countries), "country_code")

SELECT /*+ BROADCAST(c) */ o.*, c.name
FROM orders o
JOIN countries c ON o.country_code = c.code

Which side can be broadcast

The rule: the side that must keep unmatched rows is the one streamed, so the other side is broadcast. Broadcasting the left side of a left join would lose track of left rows with no match across executors.

When broadcasting backfires

Check what Spark chose

In explain(), look for BroadcastHashJoin with a BroadcastExchange under one side, versus SortMergeJoin with an Exchange on both sides. With AQE, the plan before execution may say SortMergeJoin and change at runtime; the SQL tab of the Spark UI shows the final plan.

Common mistakes