What is Data Skew?

How it works

A shuffle sends every row with the same key to the same partition. If 60% of a clickstream has user_id = 0 (a bot, a default value or a missing id), 60% of the rows go to one task. The other tasks finish in seconds and that one runs for minutes, often spilling to disk or running out of memory.

Detect it before fixing it. The symptom in the Spark UI is a stage whose maximum task duration or shuffle read is far above the median, often with spill on the slow tasks. Confirm with a count by key or by spark_partition_id().

Typical causes:

• NULL, empty or default keys (-1, "unknown", 0) from upstream bugs or optional fields.
• Genuinely popular entities: the biggest seller, a viral post.
• Low-cardinality keys used for joins or partitioning, or time skew such as a sale day.

Fixes, roughly in the order to try them:

1. Adaptive Query Execution splits skewed partitions of sort-merge joins automatically (on by default). It does not split skewed groupBy partitions.
2. Broadcast the small side, so the large, skewed side is never shuffled.
3. Handle hot keys separately: filter out NULL keys that cannot match, or join the hot keys on their own and union the results.
4. Salt the key: add a random number to split a hot key across several partitions, aggregate in two phases or replicate the other side of a join.
5. Fix the data upstream.

Example

Counting rows per shuffle partition after partitioning by a skewed key shows one partition holding most of the data (PySpark 4.2.0).

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.master("local[2]").appName("glossary-skew").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

events = spark.range(10000).select(
    F.when(F.col("id") % 10 < 6, F.lit(0)).otherwise(F.col("id") % 100).alias("user_id"))

print(events.groupBy("user_id").count().orderBy(F.desc("count")).first())
sizes = (events.repartition(4, "user_id")
         .groupBy(F.spark_partition_id().alias("partition")).count()
         .orderBy("partition").collect())
print([r["count"] for r in sizes])
spark.stop()

Row(user_id=0, count=6000)
[800, 7100, 1000, 1100]

Common mistakes

• Adding executors or memory to a skewed stage; the hot key still lands on one task.
• Salting a join that is not skewed, which only adds work.
• Expecting AQE to fix a skewed aggregation; it splits join partitions, not groupBy partitions.
• Ignoring NULL join keys that can never match but still pile onto one partition.