Spark Data Skew Handling
Data skew means a few keys hold a very large share of the rows. Spark splits data by key, so one overloaded partition lands on one task. Every other task finishes quickly and waits for the slow one. A job that should take five minutes can run for an hour because of a single hot key.
The Layman View
Imagine ten checkout lanes in a store. Nine customers carry two items each, and one customer pushes a cart with 5,000 items. Nine lanes close within a minute. The tenth lane keeps the whole store waiting. Skew creates that tenth lane.
Spotting Skew
- Most tasks in a stage finish in seconds while one or two run for minutes.
- The Spark UI stage page shows a huge gap between the median and maximum task time.
- The maximum shuffle read size of one task dwarfs the median.
- One executor reports memory errors while others sit idle.
Finding the Hot Keys
Count rows per key and sort the counts. A short list at the top reveals the problem.
from pyspark.sql import functions as F
events.groupBy("user_id").count().orderBy(F.col("count").desc()).show(5)
user_id count
------- ---------
NULL 48,000,000 <-- empty values pile up
u_1001 350,000
u_2044 290,000
u_0007 1,200
Null values and default placeholders such as "unknown" or 0 often form the hot key. Fix these first.
Fix 1: Filter or Separate Bad Keys
clean = events.filter(F.col("user_id").isNotNull())
orphans = events.filter(F.col("user_id").isNull())
Process the clean rows with a normal join. Handle the null rows separately or drop them.
Fix 2: Enable AQE Skew Join
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
AQE splits oversized partitions in sort-merge joins automatically. Try this fix before any manual work.
Fix 3: Broadcast the Small Side
A broadcast join needs no shuffle of the large table, so skew has nothing to attack.
from pyspark.sql.functions import broadcast result = events.join(broadcast(users), "user_id")
Fix 4: Salting
Salting adds a random number to the hot key so its rows spread across many partitions. The small side repeats once for each salt value so every salted key finds its match.
SALT = 8
# Large side: attach a random salt 0..7
big_salted = events.withColumn("salt", (F.rand() * SALT).cast("int"))
# Small side: copy every row once per salt value
salts = spark.range(SALT).withColumnRenamed("id", "salt")
small_salted = users.crossJoin(salts)
result = big_salted.join(small_salted, ["user_id", "salt"]).drop("salt")
Before salting: After salting:
u_1001 --> Partition 4 (huge) u_1001_0 --> Partition 1
u_1001_1 --> Partition 2
u_1001_2 --> Partition 3
...
u_1001_7 --> Partition 8
Fix 5: Two-Step Aggregation
A group-by on a hot key suffers the same way. Aggregate with the salt first and combine the partial results next.
partial = (events.withColumn("salt", (F.rand() * 8).cast("int"))
.groupBy("user_id", "salt").agg(F.sum("amount").alias("part")))
final = partial.groupBy("user_id").agg(F.sum("part").alias("total"))
Checking the Result
- Open the stage page in the Spark UI and compare the task time distribution.
- Confirm the maximum task time sits close to the median.
- Measure the whole job runtime before and after the change.
Choosing a Fix
| Situation | Best Fix |
|---|---|
| Null or placeholder keys dominate | Filter or separate them |
| Join of two large tables with a hot key | AQE skew join, then salting |
| One table is small | Broadcast join |
| Group-by on a hot key | Two-step aggregation |
Picking the Salt Size
A larger salt spreads rows more evenly and copies the small table more times. Start with a value near the ratio between the hottest key and the median key. A hot key with 80 times the median volume suggests a salt of around 8 to 16. Measure the gain and adjust.
Salt too small --> hot partition remains slow Salt too large --> small table explodes in size and slows the join