Data Engineering glossary
What is a Broadcast Join?
A broadcast join, or broadcast hash join, is a join strategy in which the engine copies the small table to every executor and builds an in-memory hash table from it, so each partition of the large table is joined locally. The large table is never shuffled, which removes the most expensive part of a join.
How it works
In a sort-merge join, Spark shuffles both tables so rows with the same key meet on the same executor, then sorts and merges them. Shuffling a large table is expensive. In a broadcast join, Spark collects the small side to the driver, sends a full copy to every executor, and probes it from each partition of the large side where that partition already sits.
Spark broadcasts automatically when it estimates one side is smaller than spark.sql.autoBroadcastJoinThreshold (10 MB by default; -1 disables it). You can force it with F.broadcast(df) or a BROADCAST hint, and Adaptive Query Execution can switch a planned sort-merge join to a broadcast join at run time once a shuffle stage reveals the real size.
Only some sides can be broadcast. The build side must be one whose unmatched rows Spark does not need to output, because each executor sees only its slice of the other table. Left outer, left semi and left anti joins can broadcast only the right side; a full outer join cannot use a broadcast hash join. A hint Spark cannot honour is silently ignored, so check the plan.
Broadcasting also helps with data skew: if the large side is never shuffled, its hot keys do not pile onto one task.
The danger is a “small” table that is not small. It is held in memory on the driver and on every executor, so a few hundred megabytes across a large cluster can cause out-of-memory errors, and Spark refuses to broadcast anything over 8 GB.
Example
The plan for a join with a hinted small dimension shows BroadcastHashJoin and no shuffle of the large side (PySpark 4.2.0).
from pyspark.sql import SparkSession, functions as F
spark = (SparkSession.builder.master("local[2]").appName("glossary-broadcast")
.config("spark.sql.autoBroadcastJoinThreshold", "-1").getOrCreate())
spark.sparkContext.setLogLevel("ERROR")
orders = spark.range(100000).select(F.col("id").alias("order_id"), (F.col("id") % 3).alias("country_id"))
countries = spark.createDataFrame([(0, "IN"), (1, "GB"), (2, "US")], "country_id LONG, code STRING")
joined = orders.join(F.broadcast(countries), "country_id")
plan = joined._jdf.queryExecution().executedPlan().toString()
print("BroadcastHashJoin" in plan, "SortMergeJoin" in plan)
print(joined.groupBy("code").count().orderBy("code").collect())
spark.stop()True False
[Row(code='GB', count=33333), Row(code='IN', count=33334), Row(code='US', count=33333)]Common mistakes
- Forcing a broadcast of a table that is large after filters or joins, which crashes the driver or executors.
- Hinting the wrong side of a left outer join; Spark silently ignores the hint and shuffles anyway.
- Raising the auto-broadcast threshold cluster-wide instead of hinting the specific join.
- Trusting stale table statistics that make a large table look small, or the reverse.