Execution models across machines

When the handoff costs milliseconds and can fail, how do you choose an execution model?

CaseConstructed case · Hana at Gridlark

The nightly job that restarts from zero

Two things trouble Hana.

The nightly billing job keeps dying. It has 10,000 tasks of about 6 minutes each (0.1 task-hours) and needs about 1.6 hours per attempt on the 640 cores. It runs on cheap spot machines (machines the cloud provider can take back at any moment). Four nights in a row, one task has failed late in the run, and Lark restarted the whole query from zero each time.
The serverless pilot is fast but expensive. She ran the same pipeline on a functions-as-a-service platform. It started 1,000 workers in about a second. But the invoice is dominated by storage request charges, not by compute.

Neither is a bug in the code. They come from the difference in scale between the models of Inside one worker and the models of this article.

How it ends →

The scale break

Everything in Inside one worker assumed that a boundary costs tens of cycles and cannot fail. The remaining models assume that a boundary costs milliseconds and can fail. Both assumptions change the answers, and the second changes them more.

Fixed cost. At this scale the fixed cost per control transfer is task dispatch: a scheduler round trip, serializing the task's code (a closure), starting a process or container. It is on the order of 1–100 ms. The fixed-cost principle gives the natural unit directly. For dispatch overhead under 1% at F = 10 ms, a task must run at least 10 ms ÷ 0.01 = 1 second. This is the entire justification for the guidance "tasks should run from 100 ms to a few seconds". It also explains why a fixed default shuffle-partition count is wrong at both ends: too few partitions and tasks are too coarse to balance, too many and you pay dispatch for work that is not there.

Failure. The other new variable is failure. For n tasks of duration T each, with a failure rate λ per task-hour, the expected number of failures per attempt is

A model that restarts the whole query on any task failure succeeds in an attempt only when no task fails. If failures are random and independent, the probability of zero failures is about e^(−nTλ). So the expected number of attempts is about e^(nTλ), and the model stops being practical once nTλ approaches 1. (this assumes independent failures. For example, at nTλ = 1 an attempt succeeds with probability e⁻¹ ≈ 37%, so you need about 2.7 full attempts on average; at nTλ = 3 you need about e³ ≈ 20 attempts; each extra unit multiplies the cost by 2.7.)

Worked example (Hana's billing job)

n = 10,000 tasks, T = 0.1 hour, λ = 10⁻³ per task-hour (spot machines). Then nTλ = 10,000 × 0.1 × 0.001 = 1. Whole-query restart is not a strategy. A second example with the same product: 10⁴ tasks running one hour each with a per-task-hour failure rate of 10⁻⁴ also gives nTλ = 10⁴ × 1 × 10⁻⁴ = 1.

RelationshipRestarting the whole query →

e^{- n T λ} succeeds; e^{n T λ} attempts

$n$: tasks in the query
$T$: hours each task runs
$λ$: failures per task-hour

Data table

Expected full attempts by n·T·λ
n · T · λ	Expected attempts
0	1×
0.063	1.06×
0.125	1.13×
0.188	1.21×
0.25	1.28×
0.313	1.37×
0.375	1.45×
0.438	1.55×
0.5	1.65×
0.563	1.76×
0.625	1.87×
0.688	1.99×
0.75	2.12×
0.813	2.25×
0.875	2.4×
0.938	2.55×
1	2.72×
1.06	2.89×
1.13	3.08×
1.19	3.28×
1.25	3.49×
1.31	3.72×
1.38	3.96×
1.44	4.21×
1.5	4.48×
1.56	4.77×
1.63	5.08×
1.69	5.41×
1.75	5.75×
1.81	6.13×
1.88	6.52×
1.94	6.94×
2	7.39×
2.06	7.87×
2.13	8.37×
2.19	8.91×
2.25	9.49×
2.31	10.1×
2.38	10.8×
2.44	11.4×
2.5	12.2×
2.56	13×
2.63	13.8×
2.69	14.7×
2.75	15.6×
2.81	16.7×
2.88	17.7×
2.94	18.9×
3	20.1×
3.06	21.4×
3.13	22.8×
3.19	24.2×
3.25	25.8×
3.31	27.5×
3.38	29.2×
3.44	31.1×
3.5	33.1×
3.56	35.3×
3.63	37.5×
3.69	39.9×
3.75	42.5×
3.81	45.3×
3.88	48.2×
3.94	51.3×
4	54.6×
4.06	58.1×
4.13	61.9×
4.19	65.9×
4.25	70.1×
4.31	74.6×
4.38	79.4×
4.44	84.6×
4.5	90×
4.56	95.8×
4.63	102×
4.69	109×
4.75	116×
4.81	123×
4.88	131×
4.94	139×
5	148×

Principle: Independence is assumed →

This single product (task count × duration × failure rate) is the dividing line between the pipelined MPP world (massively parallel engines that stream rows between stages and restart the whole query on failure) and the materialized DAG world (engines that write each stage's output and retry single tasks). It is why the question "should stage edges be materialized?" has different right answers for a 200 ms dashboard query and a six-hour batch job. These are not two philosophies. They are one inequality.

Dataflow / DAG execution

The model: the plan is a graph (a DAG, a directed acyclic graph) held by a scheduler. The vertices are stages. Each stage is p independent tasks. The edges between stages are either materialized (written out and stored) or streamed (passed along while running). The scheduler's job is to dispatch the tasks whose inputs exist.

What materializing an edge buys. Derived rather than assumed:

A materialized edge makes the producer's output re-readable.
That makes a consumer task retryable without re-running the producer.
That makes the unit of failure a task instead of a query.
Combined with a recorded graph, it also makes the producer itself recomputable, so nothing else needs to be checkpointed.

What it costs:

A write and a read of the full intermediate data through storage.
A barrier: the consumer cannot start before the producer finishes, so computing and communicating cannot overlap.

Streaming the edge inverts both. There is no intermediate write, compute overlaps communication, and latency to the first row collapses. But every stage must be resident at the same time, so the widest point of the plan sets your concurrency limit, and any task failure kills the query.

This is the principle "costs are moved, not removed" in a clean form. The cost of recovering from failure is not eliminated by either choice. It is relocated into I/O and latency (materialize) or into restart probability and memory residency (stream). And the choice is not a matter of taste. It is decided by the nTλ inequality above.

A clarification. "Dataflow/DAG" is not a peer of the models in Inside one worker. It is a layer above them. A DAG engine still needs an execution model inside each task, and modern ones use a vectorized push pipeline there. Spark is a materialized DAG whose tasks run a pull-at-the-edges, compiled-inside pipeline. Listing them as alternatives to Volcano is a category error.

BSP supersteps

BSP (Bulk Synchronous Parallel) runs a computation as a series of supersteps. In each superstep every worker computes on its own data, then they exchange messages, then they all wait at a global barrier before the next superstep starts.

The reason to accept this barrier is not performance. It is that BSP makes the computation deterministic and analyzable. The cost of one superstep is

with no interaction terms, so you can reason about an iterative algorithm's cost without reasoning about interleaving. For graph iteration, where each round must see a consistent snapshot of the previous round's values, that determinism is the point.

The cost is where this section sharpens the principle "variance costs more than volume." From Three idlenesses, a barrier converts variance into latency: E[max]/E[mean] ≈ 1 + c·√(2 ln n). What BSP adds is that the penalty does not average out across rounds. It multiplies by the number of rounds. A fast task in superstep k+1 cannot start early to make up for a slow one in superstep k, because the barrier forbids it. So a 2× straggler penalty over 30 supersteps costs 2× thirty times, not 2× once. If independent variance across rounds were allowed to overlap, it would average out. The barrier is exactly what forbids the averaging. This is the strongest statement of why barriers are expensive.

Correcting the folklore. "BSP is slow because of barriers" is imprecise. Barriers cost two separate things: (1) the max-versus-mean penalty, and (2) the prohibition on overlapping computation with communication. The second is often larger. Push-based or pipelined shuffle recovers it without giving up the determinism of the stage boundary. The escapes from the first are asynchronous and bounded-staleness models. They buy back the overlap, and pay in convergence guarantees and reproducibility. The parameter-server world (machine-learning systems where workers update shared model parameters) made exactly that trade and accepted non-determinism for it.

BSP is not exotic. Spark's stage model is BSP with lineage-based recovery (a lost piece of data is recomputed from the recorded steps that made it), and MapReduce is BSP with two supersteps. Treating BSP as a special graph-processing model, when it is the execution model of the most widely used batch engines, hides the connection.

Actor-model execution

The actor model is a design where each actor owns private state, has one mailbox (a queue of incoming messages), and processes one message at a time. Actors communicate only by sending messages. The constraint is chosen, not imposed: no shared mutable state.

What it buys. Exclusion idleness goes to zero by construction, because there is nothing to lock. And because interaction is by message, location becomes transparent: the same code runs across cores and across machines.