One task running for an hour while 199 sit idle
Skew is the most common reason a Spark job is slow, and it looks exactly like a job that is nearly finished.
A stage shows 199 of 200 tasks complete within two minutes, and the last one runs for an hour. The cluster is idle. The job is not.
That is data skew, and it is the single most common performance problem in distributed processing.
Where it comes from
Data is partitioned by a hash of the key. If keys are evenly distributed, partitions are even. Real keys rarely are.
A null key that dominates. A default tenant that owns most rows. A "guest" user id attached to half the events. Any of these puts a disproportionate share of the data into one partition, and one task must process all of it.
Adding executors does not help. The bottleneck is one task, and it can only run on one core.
Finding it
The stage view shows task duration distribution. Look at max against median — a max many multiples of the median is skew, and it is usually obvious once you look.
Then find the culprit keys by counting rows per key and looking at the top few. It is almost always a small number of values, and frequently one.
Fixes, in order of preference
Filter it out if it is junk. Nulls and sentinel values are often not wanted at all, and removing them before the join is the cheapest possible fix.
Broadcast the small side if the join has one. A broadcast join has no shuffle for that side, so there is no partition to be skewed.
Salt the hot keys if you genuinely need them. Append a random suffix to the key on the large side, replicate the matching rows on the small side across the same suffixes, join, then aggregate. It spreads one enormous partition across many, at the cost of some duplication and a more complex job.
Or let adaptive execution handle it — modern Spark can detect skewed partitions at runtime and split them automatically. Enabling it is cheap and it resolves a good share of cases without any code change.
Adding executors to a skewed job buys you more idle executors.
Partition count, separately
Independent of skew, the partition count matters. Too few and you cannot use the cluster; too many and per-task overhead dominates — thousands of tasks each processing a handful of rows spends its life on scheduling.
The usual guidance is a few multiples of your total core count, with partitions large enough to be worth a task. Adaptive coalescing handles the small-partition end automatically, which removes the most common version of this mistake.
Reading the plan first
As with any query engine, the plan tells you what will happen before you wait for it. Where are the exchanges? Which join strategy was chosen? Are filters pushed down to the scan?
Most of my Spark wins have come from disagreeing with one line of a plan, not from adding capacity.