Spark joins: broadcast and shuffle

To join two distributed tables, rows with the same key need to be on the same machine. There are two ways to get that.

Shuffle (sort-merge join)

Both sides are redistributed by the key (shuffle), sorted and matched. It works for any size, but it moves both sides across the network and writes shuffle files to disk. It is the default for two large sides.

Broadcast (broadcast hash join)

The small side is copied whole to each executor, and the large side stays where it is: no shuffle of it. Spark does this on its own when it estimates that one side is smaller than spark.sql.autoBroadcastJoinThreshold (default 10 MB); AQE can decide again during execution, with the real size.

from pyspark.sql import functions as F

sales = spark.table("silver.sales")       # billions of rows
stores = spark.table("silver.stores")     # a few hundred

joined = sales.join(F.broadcast(stores), "store_id")
joined.explain()

In the plan, look for BroadcastHashJoin (or BroadcastExchange) instead of SortMergeJoin with two Exchanges. In SQL, the same hint is SELECT /*+ BROADCAST(stores) */ ....

When not to broadcast

• The "small" side does not fit comfortably in the memory of each executor and of the driver (the driver builds the copy).
• The small side is the left side of a LEFT JOIN: Spark cannot broadcast the side that has to be preserved in full.

Join with skew

If one key appears in millions of rows, its partition in the sort-merge join becomes gigantic. AQE splits skewed partitions (spark.sql.adaptive.skewJoin.enabled, on by default); when that is not enough, filtering out the hot key and handling it separately, or *salting*, solves it.

• A NULL key never matches: many NULLs in a join key are rows that go to the same partition and are useless. Filter before joining.
• Joining on columns of different types (text with number) forces a conversion and can ruin the plan.