Nine hundred ninety-nine tasks finished. Why is the last one still running?

Two slow tasks can mean very different things

Imagine watching two execution timelines. In one, task 417 processes approximately the same number of records as every other task but crawls on a host with a failing disk. In the other, task 417 runs on a healthy host and has fifty times the input of the median task. Their symptoms are similar. Their fixes are not.

I would compare input records, input bytes, shuffle bytes read, spill volume, CPU time, garbage collection, and remote-fetch waiting time for the slow task against peers. Then I would compare the host's disk, CPU and network metrics. A bad host often shows abnormal processing rate for an otherwise ordinary partition. Skew shows abnormal work size, even if the worker is functioning perfectly.

Task duration by itself cannot distinguish them. Neither can overall cluster CPU utilization. A cluster can look idle because 999 tasks have finished and one partition still has most of the remaining work.

What speculative execution actually does

Speculation starts another attempt of a task that appears unusually slow. If the first attempt is stuck on a struggling machine, a copy running elsewhere may finish sooner. Spark exposes speculation settings that identify slow tasks relative to others and may relaunch them. These are configurable and do not mean the system always speculates by default. Spark configuration

Now consider a customer with 400 million rows assigned to one reduce partition. Launching an identical attempt elsewhere sends the same large workload to a second worker. The partition has not become smaller, and both attempts may compete for shuffle reads, storage bandwidth and memory. Speculation can make total resource consumption worse while the stage still waits for one full copy of that work to finish.

For our scenario, I would not start by toggling speculation. I would look at the per-task bytes and records, then identify heavy keys. Sampling can help, but sampling must be sufficient to detect rare very large groups. A high-cardinality aggregate with uniform keys may have a completely different explanation from a few huge customer groups.

Splitting a key without changing its answer

Exact counts are convenient here because partial counts can be combined. For the heavy customer, split its records across several salted subkeys such as (customer_17, bucket_0) through (customer_17, bucket_15). Compute 16 partial counts, then sum them in a second stage to get the customer's total.

The bucket must be derived from something that distributes records, not just from customer_17, or every event will still land in the same bucket. A hash of a stable event ID is one option. If duplicate events exist, the counting contract must determine whether to deduplicate before counting.

But I would not salt every customer automatically. It increases intermediate keys, shuffle work and aggregation stages. Reserve the extra partitioning for keys that actually need it, or use an adaptive mechanism that can split skewed partitions where it is supported. Spark's adaptive query execution can optimize particular skewed joins, but that does not imply it automatically parallelizes every arbitrary per-key aggregation. Spark SQL tuning

The same technique is not safe for every question. If each customer requires a strict sequence of stateful updates, processing pieces independently may violate the meaning of the computation. Even for numeric aggregation, floating-point regrouping can alter the last few bits of a result, so an exactness requirement needs a defined numeric model.

Why doubling the cluster might achieve nothing

Assume the job is blocked on one nonparallel reducer that needs 120 minutes on its current hardware. Doubling the number of idle executors does not make that reducer finish in 60 minutes. Its work must become divisible or the task must become faster.

There are legitimate reasons to add capacity. If many large partitions are waiting in the scheduler, extra slots can help. If the slow task suffers because its host is overloaded by other work, moving or isolating it can help. If the entire stage is throttled by shared storage, extra workers may instead increase pressure on that dependency.

I would test one targeted change and observe whether the tail task's processed records per second improves. A reduction in average task duration that leaves the critical tail unchanged is not a useful outcome for this incident.

One more production concern: speculative attempts and task retries can execute code more than once. A task that directly sends an email or updates an external database must not rely on the scheduler for exactly-once business effects. Batch computations should write into a controlled output commit path rather than allowing every attempt to publish irreversible side effects.

Once skew is under control, the next question is whether a join has to move both huge datasets across the cluster at all.

Your decision. The longest task has 45 times more records than the median, but its records-per-second rate is healthy. Would you enable speculation, add nodes, or change the key layout? Explain the first experiment.