Handle skew and elasticity
Most tasks in a stage finish in minutes, but the last few take an hour, and the whole stage waits for them. Three different causes produce exactly this picture, and the fix for each is useless against the others. Which one do you have?
You will end with which of three failure modes you have, and the fix that works for it.
Diagnose
Plot each task's duration against its input bytes. What do you see?
Bytes similar, durations scatteredImbalance
Bytes similar, one task far slowerA straggler
Duration grows with bytes, bytes very unequalSkew
Why
The plot is cheap to produce from per-task counters, and it can be proven wrong: if the points do not line up as one of these shapes, the diagnosis is wrong.
What you have
Imbalance
Use it when tasks have similar inputs but their durations vary randomly.
Fix: finer-grained tasks, dynamic assignment, work stealing. Will not work: splitting partitions.
A straggler
Use it when one task with a normal input runs far slower: a bad node, contention, a cold cache.
Fix: a speculative duplicate of that task on another node. Will not work: splitting partitions.
Skew
Use it when some tasks simply have far more data, usually because of hot keys.
Will not work: speculation. The duplicate is just as big.
Fix it
Known hot keys: salt them or separate them
If you know which keys are hot (a placeholder ID that faulty devices write, a giant customer): add a random suffix so each hot key spreads across k tasks, then combine the partial results; or route the heavy hitters to their own path.
Unknown hot keys: split adaptively
If you do not know them in advance: let the engine split oversized partitions at runtime. That needs shuffle output that can be read by byte range, which sort-based shuffles provide.
For a join, estimate work as a product
A hot key with L rows on one side and R on the other produces L × R work, not L + R. Join skew is worse than the input sizes suggest; replicating the smaller side's hot rows is the usual fix.
Whichever you have
Set the task size
Minimum task duration = dispatch cost ÷ the overhead you accept: about 1 second for 10 ms dispatch and 1%. Use at least 10 partitions per worker.
Scale between queries, not during one
Before building mid-query scale-out, compare remaining work ÷ current capacity with the time to acquire a machine. It rarely pays inside one query.