Taming data skew in Spark
Most Spark jobs that are mysteriously slow are not slow everywhere. They are slow in exactly one place. You watch the Spark UI and 199 of 200 tasks finish in seconds, while one task grinds on for what feels like forever. That one task is the whole job now, and the cluster is mostly sitting idle waiting for it. That is data skew, and once you have seen it a few times you start spotting it from across the room.
What skew actually is
Spark splits work by key. Whenever you join, group, or aggregate, every row for a given key has to land in the same place so it can be processed together. If your keys are evenly spread, the work spreads evenly too. If one key holds a wildly disproportionate share of the rows, the task that owns that key has to chew through all of them alone. Your job finishes when the slowest task finishes, so a single fat key can stretch a five minute job into an hour.
The usual culprits are boring, which is what makes them easy to miss. Null or empty keys that all collapse into one bucket. A default or placeholder value that stands in for "unknown". A handful of huge customers or accounts in an otherwise long tail. The one that gets me most often is the plain null key: a nullable join column that a large slice of rows never had populated, so every one of those nulls hashes to the same partition and a single task inherits millions of rows that barely matter.
Spotting it in the Spark UI
You do not guess at skew, you confirm it. Open the stage that is hanging and look at the task summary. The signal is a max that dwarfs the median: median shuffle read of a few megabytes and a max of several gigabytes, or a median task time in seconds and a max in tens of minutes. When the 75th percentile looks healthy and the max is off the chart, you are not looking at a slow cluster, you are looking at one overloaded task.
The other tell is spill. A skewed task often cannot fit its partition in memory, so it spills to disk, and in the worst case the executor dies with an out of memory error and the stage retries, which is even slower. The last time this caught me, almost every task in the stage finished in a few seconds while one kept running past the half hour mark, spilling tens of gigabytes to disk before it eventually died with an out of memory error and dragged the whole stage into a retry.
Fix one: let AQE handle it first
Before you write any clever code, check whether the platform can fix it for you. Adaptive Query Execution, on by default in modern Spark and on Databricks, can detect a skewed partition at runtime and split it into several smaller ones so the load spreads across tasks. It is the lowest effort fix and often enough on its own.
The knobs that matter are whether it is enabled and how aggressively it triggers:
spark.sql.adaptive.enabled = true
spark.sql.adaptive.skewJoin.enabled = true
spark.sql.adaptive.skewJoin.skewedPartitionFactor = 5
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes = 256mA partition is treated as skewed when it is both larger than the threshold and several times bigger than the median partition. If AQE is on and skew still hurts, it usually means the skew is in a group or aggregation rather than a join, or one key is so dominant that splitting it is not enough. That is when you reach for salting.
Fix two: salt the hot key
Salting is the manual version of what AQE does: you deliberately break one giant key into many smaller ones so the work can spread. You append a small random number to the key, do the heavy operation per salted key, then combine the partial results.
For an aggregation it is a two step roll up. First aggregate by the salted key, then aggregate again by the real key to merge the partials:
from pyspark.sql import functions as F
SALT = 16 # spread each key across 16 buckets
salted = (df
.withColumn("salt", (F.rand() * SALT).cast("int"))
.groupBy("key", "salt").agg(F.sum("amount").alias("partial")))
result = salted.groupBy("key").agg(F.sum("partial").alias("amount"))For a skewed join you salt the big side with a random bucket and replicate the small side across every bucket, so the matching rows can meet without all piling onto one task. The tradeoff is real: salting adds a shuffle and some complexity, so you only reach for it on the keys that actually hurt, not everywhere. Pick the salt factor to match how lopsided the key is: on the job I have in mind a factor of sixteen was enough to bring that one runaway task back in line with the rest, and a stage that had been crawling for the better part of an hour finished in a few minutes.
Fix three: when to just broadcast
The cleanest fix for a skewed join is to not shuffle at all. If one side is small enough to fit in memory, broadcast it to every executor and the large side never has to move. No shuffle means no skew, because skew is a property of how a shuffle distributes rows, and you just removed the shuffle.
from pyspark.sql.functions import broadcast
big.join(broadcast(small), "key")This is the right move for the very common shape of a huge fact table joined to a small dimension table. Spark will often broadcast automatically under spark.sql.autoBroadcastJoinThreshold, but it can misjudge a side's size, so an explicit hint is worth it when you know better. The limit is memory: once the small side is no longer small, broadcasting it will hurt more than the skew did.
The mental model
Skew is not a volume problem, it is a distribution problem. Adding nodes does nothing if one task owns half the rows. So the question is never "how do I make this faster", it is "how do I make the work even". AQE evens it out automatically, salting evens it out by hand, and broadcasting sidesteps it entirely by removing the shuffle. Reach for them in that order, confirm each fix in the UI rather than hoping, and keep the change scoped to the one key that is actually on fire.
Boring, even work is the whole game. A pipeline where every task does roughly the same amount is a pipeline that scales when the data grows, instead of one that quietly falls over the day one customer gets big.