Distributed Training
Splitting a training run across many devices buys compute and pays in coordination. It is needed when the data, the model or the calendar does not fit on one machine — and for most models it is neither needed nor free.
The problem, the obvious approach, and why it breaks
Every lesson starts where the work starts: someone has a problem, and the first model that comes to mind looks fine offline.
The training run takes too long, or does not fit — when does spreading it across machines actually help, and what does the coordination cost?
A search team's new ranking model trains on eight months of click logs and takes three days on the one GPU box the team owns. The product lead wants weekly retrains. Someone has already asked infrastructure for "a cluster".
Ask for eight GPUs, run the same script on all of them, and expect eight times the speed. Frameworks advertise a one-line change to make training distributed, so the work seems to be provisioning.
Eight workers give four times the throughput, not eight. Every step ends with the workers exchanging gradients, and the exchange time does not shrink as compute is added — it grows with the model size and the number of participants.
- Eight workers give four times the throughput, not eight. Every step ends with the workers exchanging gradients, and the exchange time does not shrink as compute is added — it grows with the model size and the number of participants.
- The eight-way run reaches a slightly worse validation loss than the single-GPU run at the same number of epochs. Nobody changed the learning rate; the effective batch is now eight times larger, so there are eight times fewer optimizer steps per epoch (Data Parallelism).
- On day two of the first long run, one node is preempted. The job dies; the run restarts from scratch because no one had wired Checkpointing into the new launcher. The three-day run is now a five-day run.
- The bill is eight times the GPU-hours for four times the speed, before counting the engineer-weeks spent on the launcher, the input pipeline and the debugging of a run that behaves differently on every attempt.
What is being predicted, and from what data
This domain leads with these two. A target nobody defined precisely is a label nobody can trust, and a dataset nobody can describe is a model nobody can debug.
- The surrounding system predicts, for a query and a candidate document, the probability of a click; the ranking lesson (Ranking) owns what that label means. This lesson is about producing the same weights faster, or at all.
- The target of distributed training itself is a model that is *equivalent* to the single-device one — same optimisation problem, same quality — reached in less wall-clock time or with a model too large for one device.
- One example is one (query, document, clicked) triple with a few hundred dense and sparse features; the dataset is tens of terabytes in Parquet on object storage and does not fit on one node's disk, let alone in memory.
- The model is a few hundred million parameters. In FP32 its weights, gradients and optimizer state fit on one large GPU with room to spare; the bottleneck is throughput, not memory.
- Each epoch streams the data from object storage through a loader; a single GPU spends a measurable fraction of every step waiting on input.
How it actually works
Precisely enough to predict its behaviour — not a framework API.
- Training is a loop: read a batch, compute a loss, compute gradients, update parameters. Distributing it means deciding which of those four stages is split across devices and what must be exchanged to keep the devices consistent.
- The three reasons to distribute are distinct. The data does not fit or the input pipeline is the bottleneck — split the data (Data Parallelism). The model does not fit on one device's memory — split the model (Model, Tensor & Pipeline Parallelism). The wall-clock is unacceptable — split whichever stage dominates. Each reason has a different remedy and a different communication pattern.
- Every split introduces a synchronisation point where devices must agree on a value — the averaged gradient, the activation at a layer boundary, the updated weights. Agreement between independent machines is a Distributed Systems problem, and it costs bandwidth, latency and a tolerance for one participant being slow or dead (Gradient Synchronisation).
- The speed-up is bounded by the ratio of compute time to communication time per step. When a step is a large matrix multiply on a big batch, that ratio is high and scaling is near-linear; when the model is small or the batch per device is tiny, the workers spend most of each step talking.
Four stages, three reasons to split them
A training step reads a batch, runs it through the model, computes gradients and updates the weights. Distributed training is the decision to run one or more of those stages on several devices at once, plus everything required to keep the devices consistent with each other.
The reason you are distributing decides what to split. If the data or the input pipeline is the bottleneck and the model fits, every device gets the full model and a shard of the data. If the model does not fit, the model itself is cut up. If the calendar is the bottleneck, you split whichever stage the profile says dominates — and it is not always the one people assume.
When it is needed and when it is not
The honest test is a profile of a single-device step. A GPU that waits on the data loader for a third of every step is not compute-bound and gains little from a second GPU that waits on the same loader. A model whose weights, gradients and optimizer state fit in device memory with headroom has no memory reason to split.
Most models in most companies fail this test in the useful direction: they fit, they are input-bound or tiny, and the cheapest speed-up is a faster loader or a larger single machine. Distribution is the answer to a specific constraint, not a maturity level.
Why is one device not enough?
when Step profile shows the loader, decode or the object-storage read dominating; GPU compute per step is a minority of the wall-clock.
cost Fix the input pipeline — prefetching, a faster format, local caching (Parquet on the data side). Adding GPUs multiplies the wait.
when Compute-bound single-device step; the required retrain cadence needs more steps per hour than one device produces.
cost Data parallelism. Pay in gradient exchange per step, a re-tuned learning rate for the larger effective batch, and checkpointing infrastructure.
when Weights plus gradients plus optimizer state plus activations exceed one device's memory even at batch size one.
cost Shard the optimizer state first, then tensor or pipeline parallelism. Pay in activation and weight traffic across devices and in a much harder engineering problem.
when A tabular or small model that trains in hours on one box.
cost A bigger box, a sampled dataset or patience. The coordination cost of distribution would exceed the saving.
What the cluster must keep being true
A distributed run is correct when every worker runs the same code on a disjoint shard of the same data with the same hyperparameters, and the synchronised weights are the ones a single device would have reached with the larger batch. None of those properties throws an error when violated; they produce a slightly worse model that looks like noise.
So the run needs its own assumptions checked, before and during — and the check that matters most is the one against the single-device baseline, because it is the only measurement that says whether the cluster is training the same model faster or a different model.
The distributed run optimises the same objective on the same data as the single-device run, and its per-step communication stays small relative to its per-step compute.
holds when Shards are disjoint and cover the data; every worker loads identical code, config and initial weights; the learning-rate schedule has been re-tuned for the effective batch; the interconnect is dedicated and the model is large enough per step to amortise the exchange.
breaks when A sharding bug duplicates or drops data; a worker starts from a different seed or config; someone shrinks the per-device batch to fit a larger model and the compute-to-communication ratio collapses; the interconnect is shared with another job.
respond Fix the shard or config first. Then re-tune batch and learning rate. Only then ask whether the interconnect or the parallelism strategy needs to change.
1def scaling_efficiency(samples_per_s_1, samples_per_s_n, n):2 # 1.0 is perfect linear scaling; the gap is communication + stragglers3 return samples_per_s_n / (n * samples_per_s_1)4 5def cost_per_epoch_hours(samples, samples_per_s_n, n, price_per_gpu_hour):6 hours = samples / samples_per_s_n / 36007 return hours * n * price_per_gpu_hour8 9# The decision is not "is N faster" but "is N faster per dollar than 1",10# and whether the calendar constraint justifies paying the difference.The efficiency number only means something if samples_per_s_1 was measured on the same data pipeline. Measuring the single-device rate on a local disk and the cluster rate against object storage confounds the input pipeline with the parallelism.
How to build it
Most important first.
- Establish that one device is actually the bottleneck. Profile a single-GPU step: if the GPU waits on the data loader, the fix is the input pipeline, not more GPUs. If the model fits and the step is compute-bound, measure how many steps per hour one device does and what the retrain cadence requires.
- Pick the parallelism from the constraint. Throughput on a model that fits → data parallelism first. A model whose weights, gradients and optimizer state exceed one device's memory → sharded optimizer state, then tensor or pipeline parallelism, in that order of complexity.
- Treat the effective batch size and learning rate as one decision. Scaling the batch by the worker count changes the optimisation problem; the learning rate schedule must be revisited and the result compared against the single-device baseline at equal epochs (Batch Size and Learning Rate).
- Wire checkpointing and resumption before the first long run, not after the first preemption. A multi-node run has many more ways to die than a single-node one, and its cost per hour makes restarts expensive.
- Budget the whole run — GPU-hours times the number of experiments you will actually launch — before committing (Training Cost). Distributed training multiplies the cost of every hyperparameter sweep by the worker count.
What to measure
Which number actually maps to the decision — and which numbers look relevant and are not.
- Scaling efficiency: samples per second on N workers divided by N times samples per second on one worker. The number that says whether the cluster is worth its bill; below roughly half, the communication is winning.
- Time per step broken into compute, communication and input wait, per worker. The decomposition says which parallelism to change; the total does not.
- Validation loss at equal epochs against the single-device baseline. Wall-clock speed-up on a run that converges to a worse model is not a speed-up.
- Do not measure "GPU utilisation" and call it efficiency. A GPU spinning on a blocking collective operation reports high utilisation while doing no useful work.
What must stay true after deployment
The field this whole domain exists for. A model is a set of assumptions with weights attached; these are the ones a monitor or a test should be checking.
- The per-step compute time on each worker remains large relative to the per-step communication time — the model, batch and interconnect stay in the regime where scaling pays.
- All workers see the same code, the same hyperparameters and disjoint shards of the same data on every run; a misconfigured worker silently training on duplicated or missing data would look like a slightly worse model, not an error.
- The cluster stays healthy for the duration of a run, or the run can resume from a recent checkpoint without changing the schedule it would have followed.
- Offline: run the single-device baseline and the distributed run on a fixed subset for a fixed number of epochs and compare validation curves. A gap is a batch-size and learning-rate problem before it is anything else.
- Before launch: kill a worker deliberately during a short run and confirm the job resumes from checkpoint with the step counter, learning-rate schedule and data position intact.
- Over time: track scaling efficiency per run. It degrades when the model grows, the interconnect is shared with another team, or someone shrinks the per-device batch to fit a larger model.
What can go wrong
- The input pipeline was the bottleneck all along, and eight GPUs now wait on the same object-storage read path eight times as hard.
- One slow worker — a degraded GPU, a noisy neighbour, a node on a slower network link — throttles every synchronous step to its pace (Gradient Synchronisation).
- Non-deterministic reduction order makes each run's loss curve slightly different, so a regression in the model code is indistinguishable from run-to-run noise until someone runs it several times (Reproducibility).
- The distributed launcher, the checkpoint format and the sharding scheme become infrastructure that one person understands, and that person leaves.
- Coordination cost is paid every step. Adding workers is buying throughput with bandwidth, and the exchange rate worsens as the model grows relative to the batch.
- A distributed run is a different optimisation problem from the single-device one. The tuned learning rate, warmup and schedule do not transfer automatically.
- Reproducibility drops from bit-exact to statistical. Debugging a model regression now needs several runs to distinguish signal from reduction-order noise.
- The operational surface — launcher, checkpointing, resumption, node health — is a small distributed system the team now owns.
- "Eight GPUs will make it eight times faster." They will make the compute eight times faster and the communication no faster. The speed-up is the ratio, and for a small model on a slow interconnect it can be close to one.
- "Our tabular model trains slowly, so we should distribute it." A gradient-boosted model on a few hundred million rows is usually bottlenecked on a single machine's memory and I/O, and the fix is a bigger box or a smaller dataset. Multi-node tree training exists, and it is rarely the first thing to reach for.
- "The framework handles distribution, so it is a one-line change." The line is real. The changed batch size, the checkpointing, the input sharding and the straggler handling are not in it.
Where this applies
ML advice is stated as universal far more often than it is. These labels say what each claim is specific to — and where CONTESTED appears, the note gives the strongest form of the opposing view.
- SCALE-SPECIFICThe lesson applies once a model or dataset genuinely exceeds one device; below that — the overwhelming majority of tabular, small-vision and classical models — the right answer is a larger single machine, a better input pipeline or a smaller dataset, and distribution only adds cost.
- MODEL-SPECIFICLarge neural networks trained by gradient descent parallelise along the data, model and pipeline axes described here; tree ensembles parallelise by histogram construction over feature columns and have different bottlenecks, and this module does not cover them.
- CONTESTEDA serious position holds that for anything short of foundation-model pretraining a team should buy the largest single accelerator it can and never distribute: one device has no synchronisation, no stragglers, bit-exact reproducibility and one failure domain, and the gap to a multi-node run closes with each hardware generation. The counter is that a weekly retrain cadence on a large dataset is a hard product constraint that one device cannot meet at any price.
Where the depth lives
This domain teaches the model and hands the rest off by name.