The question this answers
When does splitting a computation across machines actually make it faster, and what does the split cost?
A correct result *if* the work partitions into tasks whose results combine without further communication, and if every task either completes or is re-executed. There is no guarantee of a speedup: total time is bounded below by the serial fraction plus the communication the split introduces, and that second term is what distinguishes this from single-machine parallelism.
Everything below is bought to hold this sentence. "Strongly consistent" with no scope attached is a slogan, not a guarantee — read what it actually covers, and what it explicitly does not.
A worker knows the input partition it was assigned and its own progress through it. It does not know whether its peers are ahead, behind, or dead; it does not know whether the coordinator still considers it alive; and it does not know whether its own output has been consumed. Everything it believes about the job as a whole is something the coordinator told it at some point in the past.
A node knows its own state and the messages that arrived. Everything else is inference from evidence that was already stale. "B has not replied in five seconds" is knowledge; "B is down" is a decision — and usually the bug.
The three questions any distributed computation answers
Strip any distributed compute framework — batch, stream, training, or a hand-rolled fan-out — and it answers the same three questions. How is the work split? Usually by partitioning the input, so each task owns a disjoint slice. How do partial results combine? Either they do not need to, or they combine through an associative operation, or they require redistribution — and that third case is the expensive one. Who runs what, and what happens when a worker dies? That is Who Runs What, and What Happens When a Worker Goes Quiet, and it is why a distributed job is an exercise in this domain rather than in algorithms.
The taxonomy that matters most is the second question, because it decides whether you have a cheap job or an expensive one. Embarrassingly parallel work — resize a million images, score a million rows — needs no communication between tasks at all, and scales close to linearly until you run out of machines or input bandwidth. Reducible work — count, sum, max, distinct-approximate — needs communication proportional to the number of tasks rather than the size of the data, because each task can pre-aggregate locally and send a small summary. Redistributive work — group-by, join, sort — needs every task to send data to every other task, and that is The Shuffle Is the Job, which is where the money goes.
This is why the first question to ask of any distributed job is not "how many machines?" but "which of those three shapes is it, and can I move it into a cheaper shape?" Converting a redistributive job into a reducible one — by pre-aggregating, by co-partitioning inputs so the join is local, by broadcasting a small side — is routinely a bigger win than any amount of extra hardware.
| Shape | Communication | Scales with machines? | Example |
|---|---|---|---|
| Embarrassingly paralleltypical | None between tasks | Nearly linearly, until input bandwidth binds | Transcode 10M images |
| Reducible (associative)protocol | O(tasks) — small summaries | Well; combine tree keeps it cheap | Count rows, sum revenue, HyperLogLog distinct |
| Redistributiveprotocol | O(data) across O(N×M) connections | Poorly past a point; the network binds | Group by user, join two large tables, global sort |
| Iterative with synctypical | O(data) per iteration, plus a barrier each time | Bounded by the slowest worker per round | Gradient descent with synchronous updates |
What the machine boundary adds that cores do not
The Concurrency domain already taught the shape of parallel speedup: a serial fraction caps it, coordination overhead eats into it, and eight cores give you four and a half. All of that still applies. Distribution adds three things on top, and each one is a category rather than a constant factor.
Communication is now explicit and expensive. Two threads share an address space; passing a gigabyte between them is a pointer. Two machines pass a gigabyte over a network at a rate several orders of magnitude below memory bandwidth, and the difference is why an algorithm that was obviously right on one machine can be obviously wrong across ten.
Every task can fail independently. A thread that dies takes the process with it, which is at least unambiguous. A worker that stops responding gives you A Timeout Tells You Nothing About Whether It Happened: the task may be dead, may be slow, may have finished and failed to report. The scheduler must act anyway, and acting wrongly means either a lost task or a duplicated one.
There is no shared memory and no shared clock. Every piece of coordination — "this partition is done", "here is the current model", "stop, we have converged" — is a message with all the properties messages have. No Shared Memory: Every Node Sees a Copy is not an inconvenience here; it is the reason the frameworks look the way they do.
The honest first question: does this need to be distributed at all?
A very large fraction of distributed jobs exist because the data felt big, not because it was. A single modern machine can hold hundreds of gigabytes of RAM and read from local NVMe at gigabytes per second. A well-written single-process program over 200 GB frequently beats a ten-machine cluster over the same data, because the cluster spends its time on serialisation, network transfer and scheduling that the single machine never pays.
The threshold questions are concrete. Does the working set fit on one machine, allowing for the largest instance you can rent rather than the one you happen to have? Is the job actually compute-bound, or is it waiting on I/O that distribution will not help? Is the runtime a problem for anyone, or is it a nightly job with eight hours of slack? Would a better algorithm or a columnar format cut the work by a factor that makes the question moot?
And the cost that is easiest to forget: a distributed job is an *operational* object. It has a scheduler to run, failures to interpret, a shuffle to tune, stragglers to chase, and a debugging story that is much worse than a stack trace. When Not to Distribute is the general form of this argument, and it applies with unusual force to compute, because the single-machine alternative is so much better than people expect.
- Compare against the largest single machine you can rent, not the one you have.
- Distribution helps compute-bound work; it does not help work that is waiting.
- Serialisation and network transfer are costs the single-machine version never pays at all.
- A better file format or algorithm often removes the need entirely, and costs nothing to operate.
- The debugging story for a distributed job is qualitatively worse — budget for it as a real cost.
Where the time actually goes
When a distributed job is slower than expected, the cause is almost never the user code. The recurring culprits, roughly in order of frequency: shuffle (moving intermediate data), stragglers (one slow task behind a barrier), skew (one partition holding a disproportionate share of the data), scheduling delay (tasks waiting for capacity rather than running), and serialisation (turning objects into bytes and back, which for some formats costs more than the computation it enables).
Notice how many of these are distribution artefacts rather than properties of the computation. That is the module’s thesis: the interesting cost of distributed computation is what distribution added. The four lessons that follow take the largest of those costs one at a time — The Shuffle Is the Job, Move the Computation to the Data, Who Runs What, and What Happens When a Worker Goes Quiet and One Slow Task Sets the Pace for Everything — because each has a distinct diagnosis and a distinct fix.
stage 1 (map) 1,200 tasks median 4s max 6s wall 47s
shuffle write wall 41s <-- serialising + spilling
shuffle read wall 118s <-- N x M transfer
stage 2 (reduce) 200 tasks median 3s max 96s wall 101s <-- one straggler task
------------
user code ~ 18s of 307s wall. Everything else is distribution.Key points
- Distributed compute answers three questions: how work splits, how results combine, and who runs what when a worker dies.
- Work is embarrassingly parallel, reducible, or redistributive — and the third shape is where nearly all the cost is.
- Moving a job into a cheaper shape beats adding machines, usually by a wide margin.
- The machine boundary adds explicit communication cost, independent task failure, and no shared clock or memory.
- One large machine beats a small cluster far more often than teams expect, and costs a fraction as much to operate.
- When a job is slow, the cause is usually shuffle, stragglers, skew or scheduling — not the user code.
The chain, answered
Every field here is required, which is why no lesson in this domain can recommend a design without naming what an operator sees when it fails, what survives the partition, what repairs it afterwards, and the simpler thing to consider first.
- • The input is partitioned into units of work, ideally aligned with how the data is already stored.
- • A scheduler assigns tasks to workers, preferring workers that already hold the relevant input.
- • Each worker computes over its partition locally, producing either a final result or intermediate data.
- • If results must be combined across partitions, intermediate data is redistributed — the shuffle — so that all values needing to meet end up on the same worker.
- • A combining step produces the final output, often as a tree so that no single node receives everything.
- • Throughout, the scheduler tracks liveness, re-executes tasks it believes were lost, and decides when the job is complete.
- • A worker dies mid-task and the work must be re-executed somewhere else.
- • A worker is merely slow, is presumed dead, and the task runs twice.
- • Intermediate data is lost with the worker that produced it, forcing upstream re-execution rather than just the failed task.
- • One partition is far larger than the others, so the job runs at the speed of that partition.
- • The network saturates during redistribution and every task slows at once.
- • The scheduler itself becomes unavailable, and running tasks continue while nothing new is assigned.
- • The job that never finishes: 999 of 1,000 tasks completed in minutes and one has been running for hours. Cluster CPU is near idle and the job is not making progress.
- • Retry loop with no error: a worker with a bad disk fails every task assigned to it; the scheduler retries them elsewhere and they succeed, so the job completes but takes twice as long and nothing was ever reported as failed.
- • Duplicate side effects: a task that writes to an external system is re-executed after a false failure, and the operator finds duplicated rows or duplicate notifications with no corresponding error.
- • Cascading re-execution: one worker is lost and the framework must recompute not only its tasks but every upstream task whose intermediate output lived on it. A single machine loss becomes a job-length recompute.
- • Cluster-wide slowdown at a stage boundary: every task’s latency rises simultaneously when a stage transitions, because the network is saturated by redistribution — the diagnosis is the stage boundary, not any individual task.
- • Assignment: the scheduler is a coordination point, and its availability bounds the job’s ability to make progress on new work.
- • Stage barriers: if a stage must complete before the next begins, every worker waits for the slowest, which is the core of One Slow Task Sets the Pace for Everything.
- • Completion: deciding a task is finished requires the worker to report and the scheduler to believe it — an ordinary A Timeout Tells You Nothing About Whether It Happened problem with duplicate execution as the failure.
- • The cheapest distributed jobs are the ones with the fewest barriers; each barrier converts the whole stage into a fan-out with the slowest task setting the pace.
- • Deterministic, side-effect-free tasks can be re-executed freely, and the framework’s entire fault-tolerance story rests on that property.
- • Tasks with external side effects are re-executed just as freely, and the guarantee silently becomes at-least-once — which is Where You Put the Acknowledgement Decides Everything arriving in a compute framework.
- • Loss of intermediate data escalates a task failure into an upstream recompute, so the blast radius of a machine loss depends on where materialised output lives.
- • A scheduler outage stops new assignment but need not stop in-flight tasks; whether the job survives depends on whether the scheduler holds durable job state.
- • Detect: task-level progress, not job-level progress. A job at 99% for an hour is one task, and only per-task metrics say which.
- • Contain: cap retries per task and blacklist a worker that fails repeatedly, or a single bad disk will absorb retries indefinitely.
- • Recover: re-execute lost tasks; where intermediate data was lost with the worker, recompute the upstream tasks that produced it.
- • Reconcile: for tasks with external effects, make the effect idempotent by construction — write to a task-specific location and publish atomically at the end, rather than appending as you go.
- • Verify: compare output row counts or checksums against an expectation, because a job that "succeeded" with a silently skipped partition looks identical to one that did not.
- • Per-stage wall time split into compute, shuffle write, and shuffle read — this single breakdown diagnoses most slow jobs immediately.
- • Task duration distribution per stage, especially max versus median. A high ratio is a straggler or skew, and they have different fixes.
- • Input bytes per task, which distinguishes skew (uneven data) from a slow worker (even data, uneven time).
- • Task retry counts grouped by worker — a single machine dominating retries is a hardware problem wearing a scheduling costume.
- • Time tasks spend waiting for capacity versus running, which separates a scheduling problem from a compute problem.
- • Data that genuinely does not fit on one machine, or that is already stored across many machines.
- • Embarrassingly parallel work with a real deadline, where machine count converts directly into wall-clock time.
- • Workloads that need to survive machine failures without restarting from the beginning — the re-execution model is worth a lot here.
- • Data that fits in one machine’s memory, where the cluster is slower and much harder to operate.
- • Iterative algorithms with a synchronisation barrier per iteration and small per-iteration work — communication dominates and more machines make it worse.
- • Low-latency work: framework startup, scheduling and shuffle add seconds of fixed overhead that no amount of tuning removes.
- • Anything with non-idempotent side effects per task, where the re-execution model is fighting the workload rather than helping it.
- • One big machine plus a good single-process implementation — frequently faster, always simpler to debug.
- • A database that already partitions and parallelises internally: pushing an aggregation into the store beats extracting the data and aggregating it yourself.
- • A better algorithm or file format. Columnar storage with predicate pushdown regularly reduces the work by more than a cluster would.
- • Incremental processing: compute only the new data rather than reprocessing the whole set on every run.
- • A simple task queue with independent workers, when the work is embarrassingly parallel and you do not need a framework’s fault tolerance.
Three shapes of distributed work, and which one the network eats
stage 1 (map) 1,200 tasks median 4s max 6s wall 47s
shuffle write wall 41s <-- serialising + spilling
shuffle read wall 118s <-- N x M transfer
stage 2 (reduce) 200 tasks median 3s max 96s wall 101s <-- one straggler task
------------
user code ~ 18s of 307s wall. Everything else is distribution.| Communication | Scales with machines? | Example | |
|---|---|---|---|
| Embarrassingly paralleltypical | None between tasks | Nearly linearly, until input bandwidth binds | Transcode 10M images |
| Reducible (associative)protocol | O(tasks) — small summaries | Well; combine tree keeps it cheap | Count rows, sum revenue, HyperLogLog distinct |
| Redistributiveprotocol | O(data) across O(N×M) connections | Poorly past a point; the network binds | Group by user, join two large tables, global sort |
| Iterative with synctypical | O(data) per iteration, plus a barrier each time | Bounded by the slowest worker per round | Gradient descent with synchronous updates |
What people believe, and what is true
More machines means proportionally faster.
Only for work that needs no communication. Once tasks must exchange data, added machines increase communication and can make the job slower.
Distributed compute is just parallelism with more workers.
It adds explicit communication cost, independent failure per task, and no shared clock — each of which changes what algorithm is correct, not just how fast it runs.
The framework handles failure, so failures are free.
It handles failure by re-executing, which costs time and duplicates any external side effect the task performs.
Big data needs a cluster.
Hundreds of gigabytes fit in one machine’s memory. The threshold is much higher than the reflex suggests, and the single-machine version has no shuffle at all.
Go deeper
Only the levels this lesson can honestly fill — a missing level is a claim nobody had.
Overview
Split the work, compute in parallel, combine the results. Whether that is fast depends almost entirely on how much data has to move between machines to combine them.
Practical
Classify the job first: embarrassingly parallel, reducible, or redistributive. Then check whether one machine would do. If you do distribute, read the per-stage breakdown of compute versus shuffle before tuning anything, and watch max-versus-median task duration for skew and stragglers.
Advanced
The right mental model is that distribution converts a compute problem into a communication problem, and the whole craft is minimising the converted part. Pre-aggregate so summaries move instead of rows; co-partition inputs so joins happen locally; broadcast a small side so a large side never moves; pick formats whose serialisation is not the dominant cost. Every one of those is the same move — turn a redistributive shape into a reducible one — and it is worth more than any scheduling or hardware change you can make.
Apply it
- 💬 When is adding machines to a job actively harmful?
- 💬 A job takes five minutes on a cluster and ninety seconds on your laptop over the same data. Explain how that is possible.
- 💬 Classify these three jobs by communication shape: resize images, count distinct users, join two 500 GB tables.
- 💬 What does a framework need from your task code to make its fault tolerance work?