Why is my Spark job stuck on the last task (data skew)?

Summary

• Spark normally finishes one stage before it starts the next, so a stage waits for its slowest task. When one task gets a far larger share of the data than its peers, the job looks stuck on the last task.
• Confirm skew on the stage page of the Spark UI: compare the maximum with the median or 75th percentile in the summary metrics, and sort the tasks table by duration or shuffle read.
• Adaptive query execution splits skewed partitions in sort-merge joins and has been on by default since Spark 3.2.0. The Spark docs describe the feature for joins only.
• Other fixes: broadcast the small side of a join, salt a hot join key, and handle NULL or hot keys separately.
• More partitions do not split one hot key, and speculation re-launches a slow task on the same data.

Why is my Spark job stuck on the last task (data skew)?
Spark normally finishes one stage before it starts the next, so a stage waits for its slowest task. When one task receives a far larger share of the data than the others, the Spark UI shows almost every task complete while one keeps running. That pattern is data skew.

How do I confirm it is skew?

Open http://<driver-node>:4040, go to the Stages tab and click the stage description to open its detail page. The summary metrics for all tasks appear as a table and a timeline, and the tasks table breaks the same numbers down per task. The monitoring guide notes that these tables are sortable, which makes it easy to identify slow tasks and data skew.
Compare the maximum with the median and the 75th percentile for task duration and Shuffle Read Size / Records. By default, AQE calls a partition skewed when it is larger than 5.0 times the median partition size and also larger than 256MB. The Databricks Spark UI guide compares the maximum duration with the 75th percentile and says you may have skew when the maximum is at least 1.5 times that value. Then count rows per key:

from pyspark.sql import functions as F

df.groupBy("key").count().orderBy(F.desc("count")).show(10)

What are the usual causes and fixes?

Situation	What to try
A few keys dominate a sort-merge join	Adaptive query execution skew join, on by default
One side of the join is small	Broadcast hint, which sends that side to all worker nodes
A few keys dominate a join that AQE does not cover	Salt the join key
Skewed groups in a group aggregate pandas UDF	The docs say there is no partial aggregation and a group's data is loaded into memory, so reduce group size or aggregate with built-in functions first
Many NULL values in a grouping column	GROUP BY and DISTINCT put all NULLs in one bucket, so process NULL rows separately
Output partitions of very uneven size	REBALANCE hint, which splits skewed partitions when AQE is on
One task is slow although its data is normal	spark.speculation (default false) re-launches slow tasks

How does adaptive query execution help?

AQE has been enabled by default since Spark 3.2.0. Its skew join feature "dynamically handles skew in sort-merge join by splitting (and replicating if needed) skewed tasks into roughly evenly sized tasks" when spark.sql.adaptive.enabled and spark.sql.adaptive.skewJoin.enabled are both true. The docs say the threshold should ideally exceed spark.sql.adaptive.advisoryPartitionSizeInBytes (64 MB), so you can lower it to catch smaller skewed partitions:

spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5.0")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "128m")

The factor line shows the default value, and the threshold default is 256MB. The Spark docs describe this feature for sort-merge joins, so other kinds of skew need a different fix.

How do I broadcast the small side?

The hint example from the Spark tuning guide marks the smaller table:

big.join(small.hint("broadcast"), "key")

The BROADCAST hint broadcasts the hinted side regardless of spark.sql.autoBroadcastJoinThreshold (default 10 MB). The Databricks AQE page notes that the left relation of a LEFT OUTER JOIN cannot be broadcast.

How do I salt a hot join key?

Salting spreads one key across several partitions. The large side gets a random bucket, and the other side is replicated once per bucket:

N = 8  # number of salt buckets

big_s = big.withColumn("salt", (F.rand() * N).cast("int"))
small_s = small.select("*", F.explode(F.lit(list(range(N)))).alias("salt"))
joined = big_s.join(small_s, ["key", "salt"])

The small side is replicated N times, and the rand docs note the function is non-deterministic in general.

What changes on Databricks?

AQE is enabled by default. The Databricks AQE page says it handles skew in sort merge joins and shuffle hash joins and never skew-optimizes broadcast joins, and that for a LEFT OUTER JOIN it can optimize only the left side. The Skew and spill page of the Spark UI guide says to check for spill first, then look for a maximum task duration well above the 75th percentile.

FAQs

Will raising the partition count fix skew?

Not for a single hot key. When you partition by the key, as in repartition(n, "key"), the result is hash partitioned, so rows with the same key land in the same partition and more partitions cannot split that key. The default for spark.sql.shuffle.partitions is 200.

Why did AQE not detect my skew?

A partition must exceed both the threshold and the factor times the median partition size. The Spark docs describe the feature for sort-merge joins, and the Databricks docs add shuffle hash joins, so a plain groupBy or a broadcast join is not covered.

Is skew the only reason for a slow last task?

No. The stage page also reports GC time and shuffle spill, and the Databricks guide says the first thing to look for in a long-running stage is spill. Check those before changing the query.