Distributed Training: Data Parallel, Model Parallel and FSDP Strategies

Training a 7B parameter model on a single 80GB A100 is no longer a question of patience — it's simply impossible. Even with bfloat16 and activation checkpointing, the optimizer states alone for Adam will consume over 80GB. I hit this wall in late 2023 when trying to fine-tune a 13B model on 8xH100s: vanilla DistributedDataParallel kept OOMing despite batch size 1, while naive model parallelism left GPUs idle half the time. The reality of modern distributed training is that there is no single strategy that wins everywhere. You have to understand the tradeoffs between replicating data, sharding it, slicing the model, and sharding the model — and how frameworks like PyTorch FSDP orchestrate all of this under the hood. This article breaks down what actually happens on the wire and in memory when you choose Data Parallel, Model Parallel, or FSDP, and how to pick the right combination for your cluster.

Why Naive Data Parallelism Hits a Hard Memory Ceiling at Scale

In standard data parallel training, every GPU holds a complete replica of the model, optimizer states, and gradients. Each rank processes a different micro-batch, computes gradients locally, and then synchronizes via an all-reduce. This works brilliantly up to about 1-3B parameters on current hardware. Beyond that, the memory redundancy kills you.

Consider a 70B parameter model in mixed precision (bf16 parameters, fp32 Adam states). You need: 140GB for bf16 parameters and gradients, plus ~280GB for Adam's first and second moments in fp32, plus another 280GB for the fp32 master weights if you're using mixed precision. That's roughly 700GB of memory just for model states, replicated N times. Adding more GPUs does not reduce this per-GPU footprint at all. In my experience, teams waste weeks trying to squeeze larger models into DDP with gradient checkpointing alone before realizing they need sharding.

The Communication Cost You Pay Anyway

Even when replication fits, DDP has a communication tax. Every backward pass triggers an all-reduce of gradients. For a 7B model with bf16 gradients (14GB), an all-reduce across 32 GPUs over NVLink + InfiniBand still takes ~150-250ms per step if not overlapped. I've found that on clusters with weaker inter-node bandwidth (e.g., 100 Gbps RoCE instead of 400 Gbps IB), data parallel scaling efficiency can drop from 90% on 8 GPUs to under 60% on 64 GPUs — not because of compute, but because the network becomes the bottleneck.

PyTorch's DistributedDataParallel hides much of this complexity, but understanding its bucketed all-reduce is essential before you can reason about FSDP. If you're still tuning single-node optimization fundamentals like learning rate schedules and gradient clipping, make sure you have those dialed in first, as described in Training Deep Networks: Learning Rate Schedules, Batch Normalization and Gradient Clipping, because distributed strategies will magnify any instability.

Inside DDP: Bucketed All-Reduce, Gradient Synchronization and Overlap

DDP does not wait until the entire backward pass completes to synchronize. As soon as gradients for a bucket (default 25MB in PyTorch) are ready, it fires an asynchronous all-reduce on a separate stream. This allows communication to overlap with computation of earlier layers. The backward pass actually runs in reverse order, so buckets for the last layers are all-reduced while the first layers are still computing gradients.

import torch.distributed as dist
import torch.nn as nn
from torch.nn.parallel import DistributedDataParallel as DDP

# Initialize process group - NCCL backend is mandatory for GPU training
dist.init_process_group(backend="nccl")
torch.cuda.set_device(local_rank)

model = MyTransformer().cuda()
# Recommended: use gradient_as_bucket_view and static_graph for perf
model = DDP(model, device_ids=[local_rank], 
            gradient_as_bucket_view=True,
            static_graph=True)

optimizer = torch.optim.AdamW(model.parameters(), lr=2e-4, weight_decay=0.1)

for batch in dataloader:
    optimizer.zero_grad(set_to_none=True)
    loss = model(batch).loss
    loss.backward()  # buckets all-reduce here, overlapped
    # Gradients are now averaged across all ranks
    torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
    optimizer.step()

When DDP Still Makes Sense

I still use pure DDP for three cases: models under ~3B parameters, data-parallel fine-tuning with LoRA where the trainable parameters are tiny, and as the outer replication layer in hybrid setups. It is the simplest to debug, has the most mature tooling, and avoids the extra all-gathers that sharding introduces. The key optimization is to ensure your buckets are sized correctly. Too small and you launch thousands of tiny NCCL kernels with high latency; too large and you lose overlap. For most transformer workloads, the default 25-50MB is close to optimal on H100 clusters.

One subtle trap: DDP with find_unused_parameters=True disables static graph optimizations and forces extra synchronization. If your Transformer Architecture from Scratch: Attention, Encoders and Decoders has conditional branches or you use activation checkpointing with unused parameters, fix the architecture or wrap selectively rather than paying this tax globally.

DDP also assumes your batch can be sharded evenly. For variable-length sequences, uneven batching causes some ranks to finish backward faster and then stall at the all-reduce barrier — a classic straggler problem. Use a distributed sampler with padding-aware batching or drop the last incomplete batch.

Slicing the Model: Tensor Parallelism and Pipeline Parallelism in Practice

When data replication alone cannot fit the model, you have to slice the model itself. There are two orthogonal ways to do this, and production systems like Megatron-LM almost always combine them.

Tensor Parallelism: Sharding Within a Layer

Tensor parallelism (TP), popularized by Megatron-LM, splits individual layers horizontally or vertically across GPUs. For example, in attention, the QKV projection (hidden x 3*hidden) is sharded column-parallel: each rank holds a slice of the weight matrix and computes a slice of QKV. An all-gather or all-reduce is then needed to collect results. For the MLP, the first linear layer is column-parallel and the second is row-parallel, so you only need one all-reduce per transformer block in the forward pass and one in the backward.

This is extremely efficient within a node where NVLink provides 900GB/s bidirectional bandwidth, but painful across nodes where bandwidth drops by 5-10x. In my experience running TP=8 across two nodes, the all-reduce for every layer added 30% step time compared to TP=4 constrained to a single node. The rule I follow now: never span tensor parallelism across node boundaries unless you have InfiniBand with SHARP acceleration. Keep TP size <= GPUs per node (typically 8).

Pipeline Parallelism: Sharding Across Layers

Pipeline parallelism assigns contiguous blocks of layers to different GPUs. GPU 0 holds layers 0-7, GPU 1 holds 8-15, and so on. Activations are passed point-to-point between stages. The classic problem is the pipeline bubble: while the first micro-batch goes through stage 0, stages 1-3 are idle. The GPipe schedule and later 1F1B (one-forward-one-backward) interleaved schedules mitigate this by splitting a batch into many micro-batches and overlapping them.

Pipeline parallelism has far lower communication volume — only activations at stage boundaries, not every layer — but it introduces scheduling complexity and load balancing issues. If your model has an uneven structure (e.g., a large embedding layer plus 32 uniform transformer blocks), naive layer partitioning will leave one stage overloaded. I profile the per-layer memory and compute with torch.profiler and then assign layers to balance both, not just layer count.

Pure pipeline parallelism is rarely used alone today; it is combined with data parallelism (e.g., 4 pipeline stages x 8 data parallel replicas = 32 GPUs) or with FSDP. For a detailed framework-level comparison that affects how you implement these strategies, see PyTorch vs TensorFlow in 2026: Which Framework for Which Use Case.

FSDP Under the Hood: Sharding, Unsharding and the Move from FSDP1 to FSDP2

Fully Sharded Data Parallel is not just "DDP with sharding." It is an implementation of the ZeRO (Zero Redundancy Optimizer) stages that trades communication for memory savings by never materializing the full model on any single GPU except transiently.

Under ZeRO-3 / FSDP full shard: each GPU holds only a shard (1/N) of parameters, gradients, and optimizer states. During forward, FSDP performs an all-gather to reconstruct the full parameters for a transformer block just in time, runs forward for that block, then discards the full parameters and keeps only its shard. Backward does the same but also reduce-scatters gradients so each rank ends up with only its shard's gradient. This means per-GPU memory drops from O(model_size) to O(model_size / N), allowing a 70B model to fit on 32x80GB GPUs where DDP would need 700GB per GPU.

The cost is 1.5x the communication of DDP: one all-gather before forward, one all-gather before backward, plus one reduce-scatter after backward. For large models where you are memory-bound, this is a worthwhile trade. For small models, it's pure overhead.

FSDP1 vs FSDP2 and DTensor

The original FSDP (now FSDP1) in PyTorch uses FlatParameter and sharding via wrap policies. It works but has issues with composability, especially with torch.compile and activation checkpointing. FSDP2, released in PyTorch 2.2+ and now recommended, is built on DTensor (Distributed Tensor) and DeviceMesh.

FSDP2 is fundamentally simpler: parameters remain as DTensors sharded on a DeviceMesh, which makes it compatible with torch.compile and allows clean hybrid sharding. I've migrated two large training runs from FSDP1 to FSDP2 and saw 7-12% throughput gains purely from better overlap and compilation, plus far fewer deadlocks caused by FSDP1's resharding logic. If you are starting a new project in 2025-2026, use FSDP2 exclusively. Details are documented in the PyTorch Documentation under FSDP and DTensor.

import torch.distributed as dist
from torch.distributed.device_mesh import init_device_mesh
from torch.distributed.fsdp import fully_shard, MixedPrecisionPolicy

# FSDP2 requires a DeviceMesh to describe sharding
device_mesh = init_device_mesh("cuda", mesh_dim_names=("replicate", "shard"),
                                mesh_shape=(2, 4))  # hybrid: 2 DDP x 4 sharded

model = MyTransformer()

# Auto-wrap transformer blocks - critical for overlap
for block in model.layers:
    fully_shard(block, mesh=device_mesh, mp_policy=MixedPrecisionPolicy(
        param_dtype=torch.bfloat16, reduce_dtype=torch.float32))

fully_shard(model, mesh=device_mesh, mp_policy=MixedPrecisionPolicy(
    param_dtype=torch.bfloat16, reduce_dtype=torch.float32))

optimizer = torch.optim.AdamW(model.parameters(), lr=2e-4)

# With FSDP2, torch.compile works natively
model = torch.compile(model)

for batch in dataloader:
    loss = model(batch).loss
    loss.backward()
    optimizer.step()

Auto Wrapping, Mixed Precision and Activation Checkpointing

The most important tuning knob in FSDP is the wrapping policy. Wrapping the entire model as one flat parameter forces a single huge all-gather before any computation. Wrapping per transformer block enables block-by-block all-gather and overlap: while block N computes forward, block N+1 is being gathered on a separate stream. I always use a size-based or transformer-block auto wrap policy with blocks of ~100M-300M parameters.

Always combine FSDP with bf16 mixed precision and reduce_dtype=fp32 for stability. And when memory is still tight, add selective activation checkpointing. FSDP + checkpointing is powerful because you discard both full parameters and activations, recomputing them block by block. The throughput hit is ~15-20%, but it can be the difference between OOM and training.

Selecting and Tuning Your Strategy: A Decision Framework for Real-World Clusters

There is no universal best strategy — optimal choice depends on model size, cluster topology, and whether you are pre-training or fine-tuning. After several runs on clusters from 8 to 256 GPUs, this is the framework I use.

For models < 3B: pure DDP or FSDP with SHARD_GRAD_OP (ZeRO-2, shard optimizer only) gives near-linear scaling with minimal complexity. For 7B-13B: FSDP full shard (ZeRO-3) within a node, optionally hybrid sharded across nodes. For 30B-70B+: hybrid FSDP (e.g., shard within node, replicate across nodes) + activation checkpointing + optionally tensor parallelism for the largest layers. For 100B+: 3D parallelism: tensor parallel within node, pipeline parallel across nodes, and FSDP/data parallel as the outer dimension.

Hybrid sharding deserves special attention. Instead of sharding across all 64 GPUs globally (which would make every all-gather span all nodes), you shard within a node or within 2-4 nodes and then replicate across groups via data parallel all-reduce. This limits high-volume communication to fast intra-node NVLink and uses slower inter-node links only for the smaller outer all-reduce. Recent work tracked in arXiv Machine Learning consistently shows hybrid strategies achieve 15-30% higher MFU (Model FLOPs Utilization) on 100B-scale runs.

# Hybrid FSDP config example: reasoning about communication groups
# 32 GPUs total: 4 nodes x 8 GPUs
# Option A: Full shard across 32 (high cross-node all-gather)
# Option B: Hybrid - shard across 8 (within node), replicate across 4 nodes

from torch.distributed.fsdp import MixedPrecisionPolicy, CPUOffloadPolicy

# Hybrid: shard_group = 8, replicate_group = 4
# Results in: 8x memory reduction per GPU instead of 32x,
# but all-gather stays within NVLink domain
model = ...  # wrapped with fully_shard and DeviceMesh(4, 8) as above

# Optional: CPU offload for optimizer states if still OOM
# Trades PCIe bandwidth for HBM savings
# Fully sharded + CPU offload can fit 70B on 4x80GB, but step time ~1.6x slower
# fully_shard(model, offload_policy=CPUOffloadPolicy(pin_memory=True))

# Gradient accumulation to keep global batch size large without
# increasing memory - sync only on final micro-batch
for i, batch in enumerate(dataloader):
    is_last_microbatch = (i + 1) % grad_accum_steps == 0
    with model.no_sync() if not is_last_microbatch else torch.no_grad():
        loss = model(batch).loss / grad_accum_steps
        loss.backward()
    if is_last_microbatch:
        optimizer.step()
        optimizer.zero_grad(set_to_none=True)

Practical Tuning Checklist From Failed Runs

First, measure Model FLOPs Utilization, not just tokens/sec. A run that looks fast but has MFU <35% on H100s is wasting money due to poor overlap or tiny batches. Second, enable NCCL_DEBUG=INFO and TORCH_DISTRIBUTED_DEBUG=DETAIL early — silent NCCL timeouts due to mismatched bucket sizes are the number one cause of hangs I've debugged. Third, always set SHARD_GRAD_OP vs full shard deliberately: if your model fits with just optimizer sharding, don't pay for full shard's extra all-gathers.

Finally, profile before guessing. One nsys or PyTorch profiler trace will show you whether you are communication-bound (long all-gather kernels overlapping nothing), memory-bound (frequent alloc retries), or compute-bound (good overlap). I've seen teams spend weeks tuning FSDP wrapping only to discover the real limiter was dataloader stalls from slow S3 reads.

Frequently Asked Questions