Broadcast Joins and Handling Data Skew

Broadcast hash join

If one side of a join is small, Spark can send a full copy to every executor and join without shuffling the large side.

from pyspark.sql import functions as F
 
sales = spark.read.parquet("s3://bucket/curated/sales/")         # large
products = spark.read.parquet("s3://bucket/curated/products/")   # small
 
joined = sales.join(F.broadcast(products), "product_id", "left")

• Spark broadcasts automatically below spark.sql.autoBroadcastJoinThreshold (default 10 MB, based on its size estimate)
• With AQE, Spark can switch a sort-merge join to a broadcast join at runtime once real sizes are known
• Do not broadcast something that is only "small" in your test data — broadcasting a large table can OOM the driver and executors

Recognising skew

In the Spark UI stage view, skew looks like 199 tasks finishing in seconds and one task running for many minutes, with a far larger shuffle read. Typical culprits: null keys, a default customer ID, one very popular product.

Fix 1 — let AQE split skewed partitions

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

Fix 2 — handle null or junk keys separately

with_key = events.filter(F.col("customer_id").isNotNull())
no_key   = events.filter(F.col("customer_id").isNull())
result = with_key.join(customers, "customer_id").unionByName(no_key, allowMissingColumns=True)

Fix 3 — salting

Spread a hot key over N sub-keys on the large side and replicate the small side N times:

N = 16
big = big.withColumn("salt", (F.rand() * N).cast("int"))
small = small.crossJoin(spark.range(N).withColumnRenamed("id", "salt"))
joined = big.join(small, ["key", "salt"]).drop("salt")

Salting costs extra data on the small side — use it for the hot keys only when AQE is not enough.