"I was a perfectly capable model on one perfectly capable machine. Then the dataset arrived, and the machine and I had a long talk about our limitations."
A Worker That Lost Its Coordinator
Modern AI is not one program on one computer; it is a distributed system whose data, computation, model, inference, and even its decision-making are spread across many machines that must communicate and coordinate to act as one. A single accelerator can hold only so many parameters, read only so many bytes per second, and answer only so many requests per second. Once any one of those ceilings is hit, the only way forward is to split the work across more machines and pay the price of moving data between them. This book is about that split: how to perform it, what it costs, and how to keep many machines behaving like one coherent intelligence. This first section sets the thesis and proves, in one short calculation, why splitting the work can be exact rather than approximate.
For most of its history, a machine learning system was a single process on a single computer. You loaded a dataset into memory, fit a model, and served predictions from the same box. That picture is still where every practitioner starts, and it is still correct for a great deal of useful work. What changed is the scale of the inputs and the models. Datasets outgrew the memory of one machine, then the disk of one machine. Models outgrew the memory of one accelerator. Request volumes outgrew the throughput of one server. Each of those ceilings, hit independently, forces the same move: stop trying to do everything on one machine, and instead distribute the work across many. The discipline of doing that move well, for artificial intelligence specifically, is what we call scale-out AI.
1. The Day One Machine Stopped Being Enough Beginner
Three independent pressures push AI off a single machine, and it helps to keep them separate because they call for different responses. The first is data. A web-scale text corpus or a video dataset is measured in terabytes or petabytes; it does not fit in the memory of one machine, and often not on the disk of one machine, so both storage and the processing of that data must be partitioned across many nodes. The second is the model. A modern foundation model has billions or trillions of parameters; the parameters alone, plus the optimizer state and activations needed to train them, exceed the memory of a single accelerator, so the model itself must be split across devices. The third is throughput. A deployed system may need to answer thousands of requests per second with a strict latency budget, more than one server can deliver, so inference must be replicated across a fleet.
These pressures are independent. A recommendation system might have modest models but enormous data and request volume. A large language model has all three at once. A research prototype might have none, in which case a single machine is the right tool and distributing the work would only add cost and complexity. Recognizing which pressures actually bind for a given system, and resisting the urge to distribute when none of them do, is the first skill this book teaches. We make the cost of distribution explicit in Chapter 3, so that "should this be distributed at all?" becomes a question you can answer with numbers.
You distribute an AI workload because a specific resource ran out: memory to hold the data, memory to hold the model, or throughput to serve the requests. Each ceiling has its own remedy (partition the data, shard the model, replicate the service), and they compose. The art is matching the remedy to the ceiling that actually binds, because every form of distribution adds communication, and communication is the tax that the rest of this book teaches you to minimize.
2. What "Scale Out" Actually Means Beginner
To scale up is to make one machine more capable: a bigger accelerator, more memory, a faster interconnect inside the box. To scale out is to add more machines and divide the work among them. The two are not rivals; real systems do both, scaling each node up to a sensible point and then scaling out across many such nodes. This book leads with scale-out, because that is where the distinctive algorithms, system designs, and failure modes of AI at scale live. Single-node efficiency matters, and we give it careful treatment in Chapter 22 as the per-node baseline that distribution multiplies, but it is never the main event. The distinction is sharp enough that Section 1.3 is devoted to it.
When we say an AI system is distributed, we mean that one or more of its essential activities has been partitioned across machines that must communicate to produce a single coherent result. The activity might be processing the data, computing the gradient, holding the model, serving the prediction, or deciding what to do next. The unifying question, asked in a different form in every chapter, is always the same: how do we split this work across machines, move the necessary information between them, and recombine it correctly, all while keeping the cost of that movement under control?
3. Data Parallelism in One Equation Intermediate
It is reasonable to fear that splitting work across machines must introduce approximation, that a distributed answer is necessarily a fuzzier version of the single-machine answer. For the most important case in this book, training by gradient descent, that fear is unfounded, and seeing why is the cleanest possible introduction to the whole subject. Consider minimizing an average loss over $N$ training examples,
$$L(w) = \frac{1}{N} \sum_{i=1}^{N} \ell(w; x_i, y_i), \qquad \nabla L(w) = \frac{1}{N} \sum_{i=1}^{N} \nabla \ell(w; x_i, y_i).$$The gradient is an average over examples, and an average decomposes. Split the $N$ examples into $K$ disjoint shards, give shard $k$ to worker $k$, and let each worker compute the sum of gradients over only its own examples. If we add up the $K$ partial sums and divide by $N$, we recover the exact full-data gradient, because addition does not care how the terms were grouped. This regrouping is the mathematical content of data parallelism, the most widely used form of distributed training, and it is exact. The combining step, summing one vector held on each worker and sharing the result, is a collective operation called all-reduce; it is so central that Chapter 4 is built around it, and Chapter 15 turns it into a training loop.
The code below makes the claim concrete on a linear-regression loss, where the gradient has a closed form. It computes the gradient the ordinary single-machine way, then again as $K$ independent shard computations combined by averaging, and reports how far apart the two answers are.
import numpy as np
rng = np.random.default_rng(0)
N, d, K = 100_000, 50, 8 # examples, features, workers
X = rng.standard_normal((N, d))
w_true = rng.standard_normal(d)
y = X @ w_true + 0.1 * rng.standard_normal(N)
w = np.zeros(d) # the point where we evaluate the gradient
# Single-machine gradient of the mean-squared-error loss.
full = (2.0 / N) * (X.T @ (X @ w - y))
# Data-parallel gradient: each worker sees ONLY its shard and returns an
# unnormalized partial sum. Summing the parts and dividing by N is all-reduce.
shards = np.array_split(np.arange(N), K)
partials = [2.0 * (X[s].T @ (X[s] @ w - y[s])) for s in shards] # K workers
allreduced = np.sum(partials, axis=0) / N # combine step
print("workers K :", K)
print("max abs difference :", f"{np.max(np.abs(allreduced - full)):.2e}")
print("relative error :", f"{np.linalg.norm(allreduced - full) / np.linalg.norm(full):.2e}")
full and the combined per-shard gradient allreduced are compared directly; the workers never see each other's data, only the final summed vector.workers K : 8
max abs difference : 3.91e-14
relative error : 6.43e-15
The difference is not small; it is zero up to the rounding error of floating-point addition. Eight workers, each blind to seven eighths of the data, jointly computed the identical gradient. That is the promise that makes distributed training worth its complexity: for this central operation, scale-out is not an approximation you tolerate but an exact reorganization you exploit. The complexity that the rest of Part III and Part IV manage is not about correctness of the math; it is about the cost and reliability of the combining step when the vectors are billions of numbers long and the workers number in the thousands.
The all-reduce you just performed by hand, summing one vector per worker and broadcasting the result, is the single most important operation in this book. It returns as gradient synchronization in data-parallel deep learning (Chapter 15), reappears as reduce-scatter and all-gather inside sharded training (Chapter 16), and becomes all-to-all when experts live on different machines (Chapter 17). Whenever you see a distributed-training method later, ask which collective it relies on; the answer is usually a relative of the one in Code 1.1.1.
In Code 1.1.1 we combined the shard gradients by hand with a sum and a division. In a real multi-machine training job you do not gather the partial gradients to one place; each worker keeps its own and the framework exchanges them directly over the network. PyTorch exposes the exact collective as a single call, and in practice you never even call it yourself, because DistributedDataParallel fires it automatically during the backward pass:
# Run with: torchrun --nproc_per_node=8 thisfile.py
import torch, torch.distributed as dist
dist.init_process_group("nccl") # join the group of K workers
local_grad = compute_shard_gradient() # this worker's partial gradient, a tensor
dist.all_reduce(local_grad, op=dist.ReduceOp.SUM) # every worker now holds the SUM
local_grad /= dist.get_world_size() # ... divide by K to get the mean
all_reduce call. Roughly a dozen lines of manual gather-and-average collapse to a single collective, and the library handles the network transport, overlap with computation, and fault signaling that Chapter 4 unpacks.Who: A machine learning engineer at a mid-size search company training a ranking model.
Situation: Nightly retraining on one high-memory GPU took eleven hours, missing the morning deployment window whenever the data grew.
Problem: The instinct was to rent the largest single accelerator available, a classic scale-up move, at roughly triple the hourly cost.
Dilemma: Scale up to one bigger box, simple but expensive and still capped by that box's limits, or scale out across eight ordinary GPUs with data-parallel training, cheaper per unit of compute but requiring a distributed training loop and a fast interconnect.
Decision: They scaled out, because the model fit comfortably in one GPU's memory; only throughput was the binding ceiling, and data parallelism addresses throughput directly.
How: They wrapped the model in DistributedDataParallel, sharded the data loader across eight workers, and launched with torchrun, a change of about thirty lines.
Result: Wall-clock training fell from eleven hours to just under two, at lower total cost than the single big GPU, and the accuracy was identical, exactly as Output 1.1.1 predicts.
Lesson: Match the remedy to the ceiling. When only throughput binds and the model already fits, scaling out beats scaling up on both speed and cost.
4. The Six Axes of Distribution Beginner
Data parallelism distributes one thing, the training computation, along one axis. A full AI system can be distributed along several axes at once, and naming them gives us the map for the entire book. Table 1.1.1 lists the six axes, the question each one answers, and the part of the book that develops it. Every later chapter is, in effect, a deep treatment of one cell of this table, and Section 1.2 walks through all six with a worked example for each.
| Axis | The question it answers | Where the book develops it |
|---|---|---|
| Distribute data | How do we process datasets too big for one machine? | Part II |
| Distribute training | How do we train across many workers? | Part III |
| Distribute the model | How do we split one model across devices? | Part IV |
| Distribute inference | How do we serve from a fleet of machines? | Part V |
| Coordinate the cluster | How do we schedule, recover, and stay efficient? | Parts I and VII |
| Distribute intelligence | How do many agents reason and coordinate? | Part VI |
The axes are not mutually exclusive; the interesting systems live where several meet. Training a large language model distributes data, training, the model, and the cluster coordination simultaneously, which is why Chapter 19 can only be written after the chapters that own each axis separately. A swarm of drones distributes intelligence and inference at the edge. Keeping the axes distinct while understanding how they compose is the conceptual spine of the book, and Section 1.8 turns the six axes into a design-space checklist you can apply to any system you meet.
5. When Distribution Helps, and When It Hurts Intermediate
Distribution is not free, and a recurring theme of this book is that adding machines can make a system slower, not faster. The combine step in Code 1.1.1 was instantaneous because the data lived in one process; across a real network, exchanging gradient vectors takes time that grows with the model size and the number of workers, and that time does not shrink no matter how fast the workers compute. If the communication per step costs more than the computation it enables, more machines buy you nothing, a trap quantified by the communication-cost models of Chapter 3 and Amdahl's law. The honest answer to "how many machines should I use?" is almost always smaller than the number you can afford, and finding it is an engineering measurement, not a slogan.
There is also the matter of failure. One machine either works or it does not. A thousand machines are, at any moment, a system in which something is probably broken: a worker has crashed, a network link is slow, a disk is full. Distribution turns reliability from a background assumption into a design problem, which is why fault tolerance threads through the entire book, from MapReduce re-execution in Chapter 6 to elastic training in Chapter 18. The benefits of scale-out and these two taxes, communication and failure, are in permanent tension, and learning to balance them is what separates a system that scales from one that merely runs on many machines.
Because communication is the tax on scale-out, a vigorous research line tries to lower it. Low-precision and structured gradient compression shrink the bytes per all-reduce; methods in the lineage of PowerSGD and 1-bit Adam report large reductions with little accuracy loss. Local-update schemes such as local SGD and its DiLoCo-style descendants (Douillard et al., 2024) let workers take several steps between communications, trading a little statistical efficiency for far less network traffic, and have been pushed toward genuinely geo-distributed and over-the-internet training. A parallel thread overlaps communication with computation so aggressively that the all-reduce nearly disappears behind the backward pass. We return to these ideas with the machinery to evaluate them in Chapter 10; for now, note that the field treats the cost of the combine step in Code 1.1.1 as a quantity to be engineered down, not accepted.
We now have the thesis (distribute the essential work across machines), the proof that one central form of it is exact (the gradient identity), and the map (the six axes). The next section makes the map usable by walking each axis in turn, naming a real system that lives on it, and showing how the axes stack inside the systems you already use every day. That tour begins in Section 1.2.
For each system, state which of the three pressures from Section 1 (data, model, throughput) forces distribution, and which axis from Table 1.1.1 you would reach for first: (a) a fraud detector that must score 50,000 card transactions per second with a tiny model; (b) a one-time research run training a 70-billion-parameter model on a corpus that fits on a single disk; (c) a pipeline that must deduplicate a 30-terabyte web crawl before any training begins. Explain why distributing along the wrong axis would not help.
Modify Code 1.1.1 so the shards have very different sizes (for example, split $N$ into one large shard and seven tiny ones). First combine the partial sums by a plain unweighted average of the per-shard mean gradients instead of summing and dividing by $N$, and measure the error against full. Then fix it with the correct size-weighted combination. Explain why unequal shard sizes break the naive average but not the all-reduce-then-divide-by-$N$ form, and what this implies for a real cluster where workers receive unequal amounts of data.
Suppose each worker holds a gradient of $P = 10^9$ floating-point numbers (4 bytes each) and the network moves data at 10 gigabytes per second. Estimate the time to communicate one gradient once, ignoring the cleverness of real all-reduce algorithms. Compare it to the time a modern accelerator needs to compute that gradient (assume a few hundred milliseconds). Without distributing anything yet, argue from these two numbers alone whether communication or computation is likely to limit a data-parallel training step, and how your answer changes if you add more workers. We make this estimate rigorous in Chapter 3.