Spark data skew in depth: diagnose skewed joins and aggregations, then fix with AQE, broadcast or salting

A stage where 199 tasks finish in 20 seconds and one runs for 40 minutes is not a cluster-size problem. Adding nodes changes nothing; the one task still processes most of the data. That is skew.

#Prove it in the Spark UI

Open the slow stage, Summary Metrics for Completed Tasks:

Also check the SQL tab: a SortMergeJoin or HashAggregate feeding the slow stage tells you which operation is skewed.

#Find the hot keys

from pyspark.sql import functions as F

(orders.groupBy("customer_id").count()
       .orderBy(F.desc("count")).limit(20).show())

# share of rows held by the top key
total = orders.count()
orders.groupBy("customer_id").count().orderBy(F.desc("count")).limit(1) \
      .withColumn("share", F.col("count") / total).show()

Common culprits: NULL or empty-string keys, a default value like -1 or UNKNOWN, a single giant tenant, or a "guest" customer.

Tip

Check NULL keys first. In an inner join they never match anyway; filter them out before the join and the skew often disappears.

#Fix 1: let AQE split skewed partitions

Spark 3's Adaptive Query Execution detects skewed shuffle partitions in sort-merge joins and splits them.

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")

A partition is treated as skewed when it is larger than both the factor times the median and the byte threshold. Lower them if AQE is not kicking in. In the SQL plan you will see skew=true on the join or an AQEShuffleRead with skewed partitions split.

Limits: AQE skew handling works for sort-merge joins (not every join type or side, for example the outer side of some outer joins) and does not fix skewed aggregations.

#Fix 2: broadcast the small side

If one side is small enough to fit in executor memory, avoid the shuffle entirely:

from pyspark.sql.functions import broadcast
result = orders.join(broadcast(customers), "customer_id")

SELECT /*+ BROADCAST(c) */ * FROM orders o JOIN customers c USING (customer_id);

Raise spark.sql.autoBroadcastJoinThreshold carefully; broadcasting a 2 GB table to every executor causes driver and executor OOMs.

#Fix 3: salting

When both sides are large, spread the hot key over N sub-keys:

N = 16
orders_s = orders.withColumn("salt", (F.rand() * N).cast("int"))
customers_s = customers.crossJoin(spark.range(N).withColumnRenamed("id", "salt"))

joined = orders_s.join(customers_s, ["customer_id", "salt"]).drop("salt")

The big side gets a random salt; the other side is replicated N times so every salt finds a match. Cost: the replicated side grows N times, so salt only the hot keys if you can:

hot = [r.customer_id for r in orders.groupBy("customer_id").count()
       .filter("count > 10000000").collect()]

orders_s = orders.withColumn("salt",
    F.when(F.col("customer_id").isin(hot), (F.rand() * N).cast("int")).otherwise(F.lit(0)))
cust_hot = customers.filter(F.col("customer_id").isin(hot)) \
    .crossJoin(spark.range(N).withColumnRenamed("id", "salt"))
cust_rest = customers.filter(~F.col("customer_id").isin(hot)).withColumn("salt", F.lit(0))
customers_s = cust_hot.unionByName(cust_rest)

#Fix 4: split the hot keys out

Process hot keys separately with a broadcast join and union the results:

hot_df = orders.filter(F.col("customer_id").isin(hot))
rest_df = orders.filter(~F.col("customer_id").isin(hot))

result = (rest_df.join(customers, "customer_id")
          .unionByName(hot_df.join(broadcast(customers.filter(F.col("customer_id").isin(hot))), "customer_id")))

#Fix 5: two-phase aggregation for skewed groupBy

partial = (events.withColumn("salt", (F.rand() * 32).cast("int"))
           .groupBy("tenant_id", "salt").agg(F.sum("bytes").alias("bytes")))
final = partial.groupBy("tenant_id").agg(F.sum("bytes").alias("bytes"))

This works for associative aggregations (sum, count, min, max). countDistinct needs approx_count_distinct or a different approach.

#Choosing

#Verify

Rerun and compare the stage's max/median duration and shuffle read. A good fix brings max close to 2-3x median. Check the total job time too: salting can make the skewed stage fast but add a costly cross join elsewhere.

#Prevent it next time

• Profile key distributions for new large joins before they reach production.
• Keep AQE on (default in Spark 3.2+) and review the skew thresholds for big jobs.
• Model data so default keys (-1, UNKNOWN) are not used as join keys.