Spark daily 1TB batch dies on one task: skewed join keys, AQE limits and salting

Our nightly enrichment job joins roughly 1 TB of the previous day's clickstream events with a user dimension and writes partitioned Parquet for the analytics team. It had run in about 40 minutes for months. Starting in late January it began taking over two hours, then failed outright three nights in a row. The driver log ended with:

26/01/29 03:41:17 ERROR YarnScheduler: Lost executor 37 on worker-14.example.com:
Container killed by YARN for exceeding physical memory limits. 30.4 GB of 30 GB physical memory used.
Consider boosting spark.executor.memoryOverhead.
26/01/29 03:41:17 WARN TaskSetManager: Lost task 1412.0 in stage 7.0 (TID 41877):
ExecutorLostFailure (executor 37 exited caused by one of the running tasks)
Reason: Container killed by YARN for exceeding physical memory limits.
...
org.apache.spark.SparkException: Job aborted due to stage failure:
Task 1412 in stage 7.0 failed 4 times, most recent failure: ...

Reading the Spark UI

The stage view told most of the story. Stage 7, the sort-merge join between events and users, had 2,000 tasks. 1,999 of them finished in under four minutes. Task 1412 ran for 50 to 60 minutes, spilled hundreds of gigabytes to disk, and was killed. The task summary metrics for shuffle read were the giveaway:

One partition received over 400 times the median. That is not a cluster sizing problem; it is data skew.

Finding the hot keys

We sampled the join key distribution directly:

from pyspark.sql import functions as F

events = spark.read.parquet("s3://example-lake/events/dt=2026-01-28/")
(events.groupBy("user_id").count()
       .orderBy(F.desc("count"))
       .limit(10)
       .show(truncate=False))

+--------------------+----------+
|user_id             |count     |
+--------------------+----------+
|null                |1183402291|
|svc-healthcheck     |  96214870|
|u_83920117          |    184220|
|u_10294411          |    171903|
...

About 38 percent of all events had a null user_id. A product release in late January had started emitting events for logged-out sessions without a user id, where previously they were dropped at the edge. The second key, an internal synthetic monitoring account, accounted for another 3 percent.

What we ruled out

• Executor memory too small. Raising spark.executor.memory from 24g to 40g let the task run longer before dying. More memory delays an OOM caused by one oversized partition; it does not fix it.
• Too few shuffle partitions. Doubling spark.sql.shuffle.partitions to 4,000 made the median task smaller and left the hot task unchanged, because every row with the same key still hashes to the same partition.
• A bad node. The failing task moved to different executors on different hosts across retries.
• AQE disabled. spark.sql.adaptive.enabled and spark.sql.adaptive.skewJoin.enabled were both true (the defaults on Spark 3.5). Yet the final plan in the SQL tab showed a plain SortMergeJoin node for stage 7, not SortMergeJoin(skew=true), so skew handling had not been applied.

Root cause

Two things combined. First, the data: all null-key rows are hashed to the same shuffle partition, so a left outer join on a column that is 38 percent null sends 38 percent of the left side to one task, even though none of those rows can ever match. Second, the plan: the join output fed directly into an aggregation grouped by user_id. Splitting the skewed partition would have broken the output partitioning that the aggregation relied on and forced an extra shuffle, and by default AQE declines a skew optimisation that introduces an extra shuffle. So the safety net we assumed was there silently did nothing.

How a shuffle join distributes rows

For a sort-merge join, Spark repartitions both sides by a hash of the join key into spark.sql.shuffle.partitions partitions, then sorts and merges each partition pair in one task. All rows with the same key land in the same partition by design; that is what makes the join correct. Null is a single value for hashing purposes, so every null key lands together. With a uniform key distribution, partitions are balanced. With one key owning a large fraction of rows, one task owns that fraction of the work and memory.

The task does not just run long. Sorting and buffering a 160 GB partition needs heavy spilling, and while execution memory is managed on heap, the off-heap side (Netty shuffle buffers, compression codecs, direct buffers) also grows. YARN measures the whole container, so the kill message mentions memoryOverhead even though the root problem is the partition size.

When AQE skew join applies, and when it does not

Adaptive query execution re-optimises after each shuffle stage using actual sizes. Its skew join rule splits a partition into several smaller ones and duplicates the matching partition of the other side, when all of these hold:

• The partition is larger than spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5.0) times the median and larger than spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (default 256MB).
• The join is a sort-merge join or shuffled hash join, and the skewed side is splittable for that join type: for a left outer join only the left side can be split.
• Applying it does not add an extra shuffle, unless spark.sql.adaptive.forceOptimizeSkewedJoin is set to true.

Our partition easily passed the size thresholds and was on the left side of a left outer join. It failed only the last condition, which is the easiest to miss because nothing in the logs says so.

Environment

Minimal reproduction

from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder
         .config("spark.sql.shuffle.partitions", "200")
         .config("spark.sql.autoBroadcastJoinThreshold", "-1")
         .getOrCreate())

n = 20_000_000
events = (spark.range(n)
          .withColumn("user_id", F.when(F.col("id") % 10 < 4, F.lit(None))
                                   .otherwise((F.col("id") % 100_000).cast("string")))
          .withColumn("amount", F.rand()))
users = (spark.range(100_000)
         .select(F.col("id").cast("string").alias("user_id"),
                 F.concat(F.lit("seg_"), (F.col("id") % 7).cast("string")).alias("segment")))

joined = events.join(users, "user_id", "left")
result = joined.groupBy("user_id").agg(F.sum("amount").alias("total"))
result.write.mode("overwrite").parquet("/tmp/skew_repro")

Forty percent null keys, broadcast disabled to force a shuffle join. In the Spark UI the join stage shows one task reading roughly 40 percent of the shuffle data.

Who is affected

Any shuffle join where a few key values dominate: nulls, default ids such as 0 or unknown, bot or service accounts, a marketplace's largest seller. Broadcast joins are immune because there is no shuffle by key, but only when the other side fits in memory, which our 60 GB dimension did not.