Distributed training

On this page 7

Distributed training is the practice of splitting the training of a machine learning model across multiple accelerators or machines, so that they jointly compute gradients and update one shared set of weights. In AI infrastructure, it is the technique that makes it possible to train models that are too large for one GPU’s memory or datasets too large to process on one device in a reasonable time.

Distributed training combines several forms of parallelism with fast interconnects and collective communication libraries. Its central engineering problem is keeping every accelerator busy with computation instead of waiting for data from its peers, because communication, not arithmetic, usually limits how far a training job can scale.

How distributed training divides the work

A training step consists of a forward pass, a backward pass that computes gradients, and an optimizer update. Distributed training partitions one or more of these dimensions: the data, the model’s layers, the tensors within a layer, or the optimizer state. The partitioning determines which devices must exchange information and how often.

The table below summarizes the principal forms of parallelism. Large language model training typically combines three or more of them, an arrangement often called 3D or 4D parallelism.

ParallelismWhat is splitMain communication
Data parallelismTraining batches across replicasGradient all-reduce each step
Fully sharded data parallelismWeights, gradients and optimizer stateAll-gather and reduce-scatter
Tensor parallelismMatrices inside each layerFrequent activations exchange
Pipeline parallelismConsecutive layers into stagesActivations between stages
Sequence or context parallelismLong sequences across devicesAttention key and value exchange
Expert parallelismMixture-of-experts layersAll-to-all token routing
Forms of parallelism in distributed training

Data parallelism is the simplest and most common starting point: each device holds a full copy of the model, processes a different slice of the batch, and the gradients are averaged with an all-reduce operation before every update. It stops working when the model plus its optimizer state no longer fits in one device’s memory, which is where sharding and model parallelism take over.

Memory and communication

Training memory is dominated not by the weights alone but by optimizer state, gradients and activations. With the Adam optimizer in mixed precision, a common estimate is about 16 bytes per parameter before activations, so a 7-billion-parameter model needs over 100 GB of memory just for training state, more than a single accelerator typically offers.

Sharding approaches, known as ZeRO stages or fully sharded data parallelism, remove this redundancy by distributing state across devices and gathering it only when needed. The trade-off is more communication in exchange for less memory, as the following table shows.

Sharding levelWhat is partitionedMemory saving
ZeRO stage 1Optimizer stateModerate
ZeRO stage 2Optimizer state and gradientsSubstantial
ZeRO stage 3 or FSDPOptimizer state, gradients and weightsLargest, scales with device count
Activation checkpointingRecomputes activations instead of storing themLarge, at about one third more compute
Sharding levels and their trade-offs

Communication speed depends on the interconnect hierarchy. Devices within one server communicate over very fast links such as NVLink, while servers communicate over InfiniBand or high-speed Ethernet with RDMA. A sound layout places the most communication-heavy parallelism, usually tensor parallelism, inside a server and the lighter data parallelism across servers. Libraries such as NCCL implement the collective operations used for these exchanges.

Synchronous and asynchronous training

Most large-scale training is synchronous: every replica finishes its step, gradients are combined, and all replicas apply the same update. This preserves the mathematics of single-device training but makes every step wait for the slowest device, so hardware stragglers directly reduce throughput.

Asynchronous schemes, historically built around parameter servers, let workers update weights without waiting for each other. They tolerate stragglers but introduce stale gradients, which can harm convergence. For modern deep learning on dedicated clusters, synchronous data parallelism with all-reduce has largely replaced parameter servers, while asynchronous and low-communication methods remain an active research area for training across loosely connected data centers.

Scaling efficiency and key metrics

Adding devices rarely produces proportional speedups. Scaling efficiency compares the actual throughput gain with the ideal linear gain, and model FLOPs utilization (MFU) expresses how much of the hardware’s theoretical arithmetic capacity the job actually uses. For large transformer training, MFU values of roughly 35 to 55 percent are generally regarded as good; values far below indicate communication or input bottlenecks.

MetricWhat it revealsWarning sign
Tokens or samples per secondOverall throughputDrops after adding nodes
Model FLOPs utilizationHardware efficiencyWell below a third
Step time varianceStragglers and network jitterIrregular spikes
Communication share of stepInterconnect bottlenecksExceeds computation time
Pipeline bubble fractionIdle time between stagesToo few micro-batches
Metrics for monitoring a distributed training job

Increasing the device count also increases the global batch size in data parallelism. Very large batches can require adjusted learning rate schedules and warmup, because convergence behavior changes beyond a critical batch size that depends on the model and data.

Reliability and common pitfalls

On a cluster of thousands of GPUs, hardware failures are routine events rather than exceptions. Long jobs therefore depend on frequent, fast checkpointing, automatic restart, and the ability to exclude faulty nodes without restarting the whole run from scratch.

SymptomLikely causeTypical fix
Job hangs at a collectiveOne rank crashed or divergedTimeouts and health checks
Out-of-memory errorsActivation peaks or poor shardingCheckpointing, smaller micro-batches
Poor scaling across nodesSlow inter-node networkMove tensor parallelism inside nodes
Loss diverges at scaleBatch too large for the scheduleLonger warmup, tuned learning rate
GPUs idle waiting for dataSlow input pipelinePrefetching and faster storage
Frequent problems and their usual causes

A frequent planning mistake is to choose parallelism before profiling. Measuring memory use and communication volume on a small configuration, then scaling one dimension at a time, almost always leads to a better layout than adopting a complex scheme from the outset.

Stripped to the essentials

Distributed training is teamwork for computers that train AI models. Some models are too big for one machine. The job is then split among many machines. They work at the same time and regularly share what they have learned.

It resembles a large group writing one encyclopedia. Each writer covers one part. They must keep in touch so the volumes fit together. The more time they spend coordinating, the less time they spend writing, which is why fast connections between machines matter so much.

Where it is applied

A research group trains a 13-billion-parameter language model on 64 GPUs spread across 8 servers, each server holding 8 GPUs connected by a fast internal link. Plain data parallelism fails immediately, because the training state does not fit in one GPU’s memory.

The engineers apply tensor parallelism across the 8 GPUs inside each server and fully sharded data parallelism across the 8 servers, with activation checkpointing to control memory peaks. Profiling shows the input pipeline starving the GPUs, so they add prefetching. Model FLOPs utilization rises to about 45 percent, and checkpoints every 30 minutes let the job recover from two node failures with less than an hour of lost work.