What is data skew and how can you mitigate it?

Detailed explanation

Symptoms: one stage where most tasks finish in seconds and a handful take minutes; large shuffle read on those tasks; spills or out-of-memory errors on a single executor.

Common causes: NULL or default keys (-1, "unknown"), a dominant customer or product, low-cardinality join keys.

Fixes

1. AQE skew join (on by default) splits oversized partitions in sort-merge and shuffled hash joins and replicates the matching partition of the other side. It does not split a skewed groupBy or window partition.
2. Broadcast the small side so the skewed large side is never shuffled.
3. Handle hot keys separately: filter NULL keys that cannot match, or process the top key in its own branch.
4. Salting: spread a hot key across N sub-keys.

Salting a join

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.master("local[2]").appName("example").getOrCreate()

N = 8
facts = spark.range(0, 50_000).withColumn("key", F.when(F.col("id") % 10 < 8, F.lit("hot")).otherwise(F.col("id").cast("string")))
dims = spark.createDataFrame([("hot", "big customer")], ["key", "label"])

salted_facts = facts.withColumn("salt", (F.rand(seed=1) * N).cast("int"))
salted_dims = dims.crossJoin(spark.range(N).withColumnRenamed("id", "salt"))

joined = salted_facts.join(salted_dims, ["key", "salt"], "left")
print(joined.filter(F.col("label").isNotNull()).count())

40000

The 40,000 “hot” rows are now spread across 8 salt values instead of one partition, and every one still finds its match because the dimension row was replicated for each salt.

Skew in aggregations

Aggregations are skewed less often than joins, because Spark already pre-aggregates on the map side: for sum, count, min or max, each input partition sends one partial row per key, however hot the key is. Skew bites when the aggregate cannot be combined early (collect_list, collect_set, exact countDistinct, percentiles) or when a window function partitions by a hot key. Then aggregate in two phases: group by (key, salt) first, then group the partial results by key.

partial = (facts.withColumn("salt", (F.rand(seed=2) * N).cast("int"))
                .groupBy("key", "salt").agg(F.collect_set("id").alias("ids")))
final = partial.groupBy("key").agg(F.size(F.flatten(F.collect_list("ids"))).alias("n_ids"))
print(final.filter(F.col("key") == "hot").first())

Row(key='hot', n_ids=40000)

The hot key’s 40,000 values are first collected in 8 smaller groups, then merged.

Common mistakes

1. Salting without confirming skew in the UI.
2. Forgetting to replicate the other side when salting a join, which drops matches.
3. Assuming AQE handles aggregation skew.