When would you use a broadcast join in Spark?

Detailed explanation

In a sort-merge join, Spark shuffles both tables so rows with the same key land on the same executor, then sorts and merges them. Shuffling a large table is expensive.

In a broadcast hash join, Spark sends a full copy of the small table to every executor and builds an in-memory hash table from it. Each partition of the large table is joined locally. The large side is not shuffled at all.

Example

from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder.master("local[2]").appName("broadcast")
         .config("spark.sql.shuffle.partitions", "4").getOrCreate())

codes = F.array(*[F.lit(c) for c in ["IN", "UK", "US", "DE"]])
orders = spark.range(0, 200_000).select(                     # the "large" side
    F.col("id").alias("order_id"),
    F.element_at(codes, (F.col("id") % 4 + 1).cast("int")).alias("country_code"))
countries = spark.createDataFrame(                           # a handful of rows
    [("IN", "India"), ("UK", "United Kingdom"), ("US", "United States"), ("DE", "Germany")],
    ["country_code", "country_name"])

joined = orders.join(F.broadcast(countries), "country_code")
joined.explain()

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [country_code#2, order_id#1L, country_name#4]
   +- BroadcastHashJoin [country_code#2], [country_code#3], Inner, BuildRight, false, false
      :- Project [id#0L AS order_id#1L, element_at(...) AS country_code#2]
      :  +- Range (0, 200000, step=1, splits=2)
      +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [plan_id=24]
         +- Filter isnotnull(country_code#3)
            +- Scan ExistingRDD[country_code#3,country_name#4]

There is no Exchange hashpartitioning above Range: the orders are joined where they are. BuildRight says the hash table was built from the right side (countries).

Spark also broadcasts automatically when it estimates a side is below spark.sql.autoBroadcastJoinThreshold (10 MB by default), and Adaptive Query Execution can switch a planned sort-merge join to a broadcast join at run time once a shuffle stage shows the real size (spark.sql.adaptive.autoBroadcastJoinThreshold, which falls back to the same value when unset).

Which side can be broadcast

The build side must be one whose unmatched rows Spark does not need to output, because each executor only sees its own slice of the other table:

spark.conf.set("spark.sql.adaptive.enabled", "false")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")   # only hints may broadcast

def strategy(df):
    plan = df._jdf.queryExecution().executedPlan().toString()
    return next(s for s in ["BroadcastHashJoin", "SortMergeJoin"] if s in plan)

print("left join, hint on right:", strategy(orders.join(F.broadcast(countries), "country_code", "left")))
print("left join, hint on left: ", strategy(F.broadcast(orders).join(countries, "country_code", "left")))
print("full join, hint on right:", strategy(orders.join(F.broadcast(countries), "country_code", "full")))

spark.conf.unset("spark.sql.adaptive.enabled")
spark.conf.unset("spark.sql.autoBroadcastJoinThreshold")

left join, hint on right: BroadcastHashJoin
left join, hint on left:  SortMergeJoin
full join, hint on right: SortMergeJoin

A hint Spark cannot honour is silently ignored, which is why you check the plan. Left outer, left semi and left anti joins can broadcast only the right side; right outer only the left; a full outer join cannot use a broadcast hash join.

When it becomes dangerous

• The “small” table is not small. It is collected to the driver, then held in memory on every executor (once per executor, not per task). A table of a few hundred MB multiplied across a large cluster, plus the driver copy, can cause out-of-memory errors. Spark also refuses to broadcast a table larger than 8 GB.
• Statistics are wrong. A filtered table may look small to the optimiser but be large in practice, or the reverse. AQE’s run-time switch uses real shuffle sizes, which is safer than static estimates.
• Slow build. Broadcasting must finish within spark.sql.broadcastTimeout (300 seconds by default); an expensive query on the small side can time out.
• Outer-join direction, as shown above.

Trade-offs

Broadcasting trades network shuffle of the large side for memory and a driver round trip for the small side. It wins when one side is small and the other large; when both are large, a sort-merge join (possibly with AQE skew handling) is the right choice, and raising the threshold just moves the failure to memory.

Common mistakes

1. Forcing a broadcast hint on a table that grows over time.
2. Raising the threshold globally to “fix” one slow job.
3. Not checking the physical plan to confirm which join actually ran.
4. Hinting the preserved side of an outer join and assuming it worked.