# Broadcast Joins and Handling Data Skew

> When to broadcast, when Spark does it for you, and three techniques for the one partition that takes an hour.

- Source: https://akshaythakare.com/notes/pyspark/broadcast-joins-and-skew
- Topic: PySpark
- Level: intermediate
- Reading time: 2 min
- Tags: pyspark, joins, performance, skew
- Updated: 2026-10-11

## 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**.

```python
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

```python
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

```python
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:

```python
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.