You kick off a Spark job, 199 of 200 tasks finish in under a minute, and the last one runs for forty minutes before failing with an out-of-memory error. That is data skew: a handful of keys own a disproportionate share of the rows, so the partitions holding them become stragglers that dictate the runtime of the whole stage.
Skew is one of the most common reasons Spark pipelines are slow and expensive, and it rarely goes away by adding more executors. This guide covers how to spot it, what Adaptive Query Execution (AQE) fixes for you automatically, and the manual techniques, broadcast joins, salting and isolating hot keys, that you still need when AQE is not enough.
What data skew actually is
Joins, aggregations and window functions all need rows with the same key to end up in the same partition. Spark does this with a shuffle: each row is hashed on its key and sent to a target partition. If the key distribution is roughly even, every task gets a similar amount of work. If one customer ID, one country code or, very often, NULL accounts for 30% of the rows, then 30% of the data lands in one task.
Typical real-world culprits include:
- Default or placeholder values such as
NULL,-1,'UNKNOWN'or an empty string in a join key. - A few very large customers, merchants or tenants in a B2B dataset.
- Bot or test traffic concentrated on a single user or session ID.
- Time-based keys where one day (Black Friday, a backfill date) is far bigger than the rest.
How to spot skew
In the Spark UI
Open the stage that is taking the longest and look at the Summary Metrics table. Compare the median and max values for Duration and Shuffle Read Size. A max that is 10x or more the median is a strong sign of skew. You will often also see high Spill (Memory) and Spill (Disk) on just one or two tasks.
In the data itself
Confirm the suspicion by counting rows per key before you start tuning. This is cheap compared with repeatedly rerunning a failing job:
from pyspark.sql import functions as F
(orders
.groupBy("customer_id")
.count()
.orderBy(F.desc("count"))
.show(20, truncate=False))
# Share of rows held by the top key vs the total
total = orders.count()
top = orders.groupBy("customer_id").count().agg(F.max("count")).first()[0]
print(f"Top key holds {top / total:.1%} of rows")If the top row is null, you may not need any clever tuning at all; see the section on hot keys below.
Let AQE do the first pass
Adaptive Query Execution re-optimises a query at runtime using the actual statistics from completed shuffle stages. It has been enabled by default since Spark 3.2, and it includes a skew join optimisation: when a shuffle partition on one side of a sort-merge join is much larger than the others, Spark splits it into smaller sub-partitions and replicates the matching partition from the other side, so the work is spread across several tasks.
The relevant settings, with their usual defaults, are:
# Core AQE switches (on by default in Spark 3.2+)
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
# A partition is treated as skewed when it is BOTH:
# larger than skewedPartitionFactor x the median partition size, AND
# larger than skewedPartitionThresholdInBytes
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")
# Target size AQE aims for when splitting or coalescing partitions
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "64MB")If your skewed partitions are, say, 180 MB, they fall under the 256 MB threshold and AQE will leave them alone. Lowering the threshold (for example to 128MB) and the factor (to 3) makes AQE more aggressive. Check the result in the SQL tab of the Spark UI: a join that AQE has rebalanced shows skew=true on the SortMergeJoin node in the final plan.
AQE also converts a sort-merge join to a broadcast join at runtime when it discovers one side is small after filtering, which sidesteps skew entirely for that join.
When AQE is not enough
AQE is good, but it has clear limits:
- Skew join handling applies to joins only. Skewed
groupByaggregations and window functions partitioned by a hot key are not split. - It only kicks in when a partition crosses both the factor and the byte threshold.
- Some join types cannot be split on certain sides. For example, the outer side of a left outer join can be split, but the side whose unmatched rows must be preserved in full has restrictions, and full outer joins offer little room.
- It cannot split a single key’s rows below the size of the matching data on the other side, so extreme many-to-many skew can still blow up.
For those cases you need to change the shape of the query yourself.
Technique 1: broadcast the small side
If one side of the join is small enough to fit comfortably in executor memory, broadcasting it removes the shuffle of the large side altogether, so there is nothing to skew. Spark broadcasts automatically below spark.sql.autoBroadcastJoinThreshold (10 MB by default), but statistics are often missing or wrong, so an explicit hint is more reliable:
from pyspark.sql.functions import broadcast
enriched = orders.join(broadcast(customers), "customer_id", "left")In Spark SQL the equivalent is SELECT /*+ BROADCAST(c) */ .... Keep broadcast tables in the tens or low hundreds of megabytes; broadcasting gigabytes moves the bottleneck to the driver and can cause out-of-memory errors there instead.
Technique 2: deal with hot keys explicitly
When the skew comes from NULL or placeholder keys, those rows usually cannot match anything on the other side. Split them off, process the rest normally, and union them back:
valid = orders.filter(F.col("customer_id").isNotNull())
missing = orders.filter(F.col("customer_id").isNull())
joined = valid.join(customers, "customer_id", "left")
# Add the customer columns as nulls so the schemas line up
for c in customers.columns:
if c != "customer_id":
missing = missing.withColumn(c, F.lit(None).cast(customers.schema[c].dataType))
result = joined.unionByName(missing)The same pattern works for a known list of genuinely hot keys: join the hot subset using a broadcast of the matching dimension rows, and join the long tail with a normal shuffle join.
Technique 3: salting
Salting spreads a hot key across several partitions by appending a random number to it. On the large, skewed side you add a random salt between 0 and N-1. On the other side you replicate each row N times, once for each salt value, so every salted key still finds its match.
SALT_BUCKETS = 16
# Large, skewed fact table: random salt per row
orders_salted = orders.withColumn(
"salt", (F.rand(seed=42) * SALT_BUCKETS).cast("int")
)
# Smaller side: explode one copy per salt value
customers_salted = customers.withColumn(
"salt", F.explode(F.sequence(F.lit(0), F.lit(SALT_BUCKETS - 1)))
)
joined = (orders_salted
.join(customers_salted, ["customer_id", "salt"], "inner")
.drop("salt"))The trade-off is that the replicated side grows N times, so pick the smallest N that fixes the straggler. A common refinement is to salt only the hot keys and leave the rest with a salt of 0, which keeps the replication cost small.
Salting aggregations
For aggregations, salting works as a two-stage aggregate: aggregate on the salted key first, then aggregate the partial results on the real key.
partial = (events
.withColumn("salt", (F.rand() * 16).cast("int"))
.groupBy("user_id", "salt")
.agg(F.count("*").alias("cnt"), F.sum("amount").alias("amt")))
final = (partial
.groupBy("user_id")
.agg(F.sum("cnt").alias("event_count"), F.sum("amt").alias("total_amount")))Spark already performs partial (map-side) aggregation for simple functions such as sum and count, so this matters most for functions that cannot be combined cheaply, such as collect_list, exact distinct counts and percentiles.
Choosing the right fix
| Situation | Best first move | Cost |
|---|---|---|
| Moderate join skew, Spark 3.2+ | Rely on AQE, tune thresholds | None, config only |
| One side under ~100 MB | Broadcast join | Driver and executor memory |
| Skew caused by NULL or placeholder keys | Split off and union back | Very low |
| A few genuinely hot keys | Isolate hot keys or salt only those | Low |
| Severe skew on large-to-large join | Full salting | Replicates one side N times |
| Skewed groupBy with non-combinable functions | Two-stage salted aggregation | An extra shuffle |
Common pitfalls
- Throwing hardware at it. More executors do not help a single straggler task; the hot partition is still processed by one core.
- Raising
spark.sql.shuffle.partitionsalone. More partitions spread the long tail, but every row for a hot key still hashes to the same partition. - Salting with a non-deterministic value in a retried stage.
rand()without a seed can produce different values if a task is recomputed. Use a seed, or derive the salt from a column such as a hash of the primary key. - Forgetting to drop the salt column. It leaks into downstream tables and breaks joins further along the pipeline.
- Disabling AQE to make plans predictable. You lose skew handling and partition coalescing. Tune it instead of switching it off.
- Fixing the symptom, not the data. If a NULL key is 40% of a fact table, that is often a data quality issue worth raising upstream.
Conclusion and key takeaways
Data skew is a data distribution problem, not a cluster sizing problem. Measure first, let AQE handle what it can, then reshape the query for the cases it cannot. If you are setting up Spark locally to experiment with these techniques, the guide on installing Apache Spark on Windows will get you started, and there is more on pipeline design on the blog.
- Compare median and max task duration and shuffle read in the Spark UI to confirm skew.
- Count rows per key; NULL and placeholder values are the most common cause.
- AQE splits skewed sort-merge join partitions automatically in Spark 3.2+, but only above both thresholds and only for joins.
- Broadcast small dimensions, isolate hot keys, and use salting as the heavier tool.
- Use two-stage aggregation for skewed groupBy operations with non-combinable functions.
