The Data Pipeline Architecture

Every public benchmark, every architecture diagram, every loss curve in a modern LLM paper sits on top of one piece of infrastructure that is almost never drawn: the data pipeline. DeepSeek-V3 was pretrained on 14.8 trillion tokens. Those tokens did not exist on disk anywhere in the world before the team built them. They were manufactured — assembled, filtered, deduplicated, scored, and repacked — from several petabytes of raw web crawl, code repositories, and scientific text. This section is the architectural blueprint of that manufacture. What goes in is web sludge; what comes out is the most information-dense corpus a 671B model has ever been fed. Everything that follows in this chapter is about doing each step well; this section is about why the steps exist, what shape they form, and how the shape governs every other engineering choice in pretraining.

The pipeline in one sentence. Take a petabyte of dirty text, run it through eight successive filters that each throw away the majority of what they see, tokenize what survives, shuffle it across thousands of GPUs, and feed it at a constant 3-million-tokens-per-second line rate for 54 days. The architecture is whatever lets that sentence finish without dropping a packet.

The Problem: From Web Sludge to 14.8T Clean Tokens

Start from the naïve question: where do the tokens come from? The public answer — "a snapshot of the internet" — hides the actual engineering problem. Raw CommonCrawl is roughly 400 TB of compressed WARC per snapshot. After HTML extraction it expands to a few petabytes of plain text. Of that, the overwhelming majority is unusable for pretraining: it is the wrong language, it is templated SEO spam, it is bit-identical copies of articles seen a hundred times in earlier shards, it is product listings, it is autogenerated forum boilerplate, it is parked-domain placeholder text, it is malware-injected mirrors. A pretraining corpus is what is left after you remove all of that. The job of the pipeline is to remove it without removing anything useful and without starving the training run.

Three pressures push back on each other and define every design choice:

1. Quality. A 671B model is unforgiving. Junk tokens do not just produce no signal — they actively burn gradient capacity that could have gone to high-information text. The Chinchilla / DeepSeek scaling laws (Chapter 9) assume tokens are roughly equal in worth, so letting a quarter of your corpus be ad copy effectively shrinks the model.
2. Volume. Even after aggressive filtering you still need $\sim 15 \times 10^{12}$ tokens of survivors. That means you must start with $\sim 10^{15}$ tokens of raw input — petabyte-scale. Per-document Python code that is too slow by 10× turns a 3-day filter pass into a 30-day filter pass.
3. Throughput. Training never pauses. The pipeline must deliver tokens to $\sim 2{,}048$ H800-class GPUs at line rate — about 3 million tokens per second aggregate — with zero stalls. A GPU starved for input is a GPU costing $10/hour to do nothing.

These three pressures form a triangle that pretraining engineers have learned to navigate by separating concerns: an offline stage that prioritizes quality and volume but can take days, and an online stage that prioritizes throughput and runs in lockstep with training. The whole pipeline is the choreography between the two.

Intuition: A Refinery, Not a Hard Drive

The instinct of someone seeing "14.8T tokens" for the first time is to picture a giant hard drive. That mental model is wrong in a useful way. The right mental model is an oil refinery.

Crude oil arrives at a refinery in vast, dirty volumes. It is fractioned in successive towers — each tower removes one class of impurity and passes the rest forward. Lighter, more valuable products come out the top; sludge comes out the bottom. The refinery's value is not the crude (which is cheap and abundant) and it is not the storage tanks (any company can buy storage). The value is the process: the sequence of towers, the temperatures, the catalysts, the throughput rate. A refinery that produces 1 million barrels a day of clean gasoline is worth more than a refinery that produces 1 million barrels a year of the same gasoline. Throughput is part of the product.

A pretraining data pipeline is the same shape. Raw crawl is the crude. Each stage — language ID, exact dedup, near-dedup, quality classifier — is a tower. The clean tokens at the bottom are the gasoline. And the rate at which the bottom of the funnel delivers tokens to the model is the refinery's output: a number that has to match the model's consumption rate every second of every training day.

Why not skip stages and just feed more crude? Two reasons. First, the downstream stages cost more per token. A quality classifier that runs at 1k docs/sec/CPU is fine when fed pre-deduplicated text; it is bankrupting when fed raw crawl because 90% of its work is on documents that exact dedup would have removed for free. The filters are ordered by cost per kept token: cheap, high-throw-away filters first; expensive, surgical filters last. Second, the model's loss responds more to the worst 5% of its training data than to the best 95%. Skipping a stage upgrades the worst 5%, not the average. The economics push you toward thorough filtering, not toward more crude.

The Math: Funnel Retention and Train-Loader Throughput

Two equations capture the entire architectural problem. The first is the funnel itself. Let the pipeline have $S$ stages, with per-stage retention rate $r_i \in (0, 1]$ — the fraction of input bytes (or documents, or tokens; we will use bytes for now and convert at the end) that survive stage $i$. Let $B_0$ be the raw input in bytes. After stage $i$ the surviving bytes are:

$B_i = B_0 \cdot \prod_{j=1}^{i} r_j$

And the final survivors in tokens are:

$N_{\text{tokens}} = \frac{B_S}{b_t} = \frac{B_0}{b_t} \cdot \prod_{j=1}^{S} r_j$

where $b_t$ is the average number of bytes per tokenized unit (about 4 for BPE on English text, lower for compressed languages, higher for code). The product $\prod r_j$ is the overall retention rate. For DeepSeek-V3-class pipelines it lands at roughly $0.5\%$ to $2\%$: you need 50–200× more raw bytes than you will keep tokens. The training token budget is the spec; the raw-input budget is the spec multiplied by $b_t / \prod r_j$.

The second equation is the throughput contract. If the model consumes $N_{\text{tokens}}$ tokens in $T_{\text{days}}$ days of training across $G$ GPUs, then the loader must deliver:

$\tau = \frac{N_{\text{tokens}}}{T_{\text{days}} \cdot 86400}$

tokens per second of aggregate throughput, and per-GPU:

$\tau_{\text{GPU}} = \frac{\tau}{G} = \frac{N_{\text{tokens}}}{G \cdot T_{\text{days}} \cdot 86400}$

For the DeepSeek-V3 numbers ($N = 14.8 \times 10^{12}$, $T = 54$, $G = 2048$), $\tau \approx 3.17 \times 10^6$ tokens/sec aggregate and $\tau_{\text{GPU}} \approx 1{,}550$ tokens/sec per GPU. At 4 bytes/token that is roughly $12.7~\text{GB/s}$ aggregate read bandwidth the loader has to sustain — every second, for two months — without ever leaving a GPU waiting.

Manual Numerical Walkthrough

Plug the DeepSeek-V3 numbers into both equations and follow them with pen and paper.

Visualizing the Pipeline

Each slice of the funnel below is one stage. The slice's left edge is the bytes entering the stage; its right edge is the bytes surviving. Drag the retention sliders on the right and watch the funnel collapse in real time. The output meter at the top right turns green when the surviving tokens cross the 14.8T target and red when they fall short.

Three things to try. First, click the "Lazy pipeline" preset. Retention is much higher per stage, the funnel barely narrows, you exceed 14.8T tokens easily — but the "tokens" you keep are saturated with templated SEO and near-duplicates. This is the trap of skipping stages: the volume is fine, the quality is not. Second, switch to the DeepSeek-V3 default and drag the near-dedup slider to 0.20. Watch the output meter dip below target — you would now need to feed 9 PB of raw input to recover. Aggressive filtering is cheap in compute and expensive in raw-input requirements. Third, increase the GPU count from 2048 to 4096 and watch per-GPU throughput halve. The loader has more slack per GPU but the aggregate bandwidth requirement still has to be met by your storage layer.

Plain Python: A Composable Generator Pipeline

Before pulling in PyTorch, it is worth writing the offline pipeline as plain Python. The whole structure is generator composition: each stage is a function that consumes an iterator and yields an iterator. The same code runs on a single laptop with a 1 MB shard and — modulo distribution — on a 4-PB CommonCrawl dump.

1from __future__ import annotations
2import hashlib
3import re
4from collections.abc import Iterator
5
6# A tiny composable pretraining pipeline. Every stage is a generator that
7# consumes (doc_id, text) tuples and yields the survivors. The whole
8# pipeline is functional composition — no framework, no DAG engine.
9
10Doc = tuple[str, str]
11
12def ingest(paths: list[str]) -> Iterator[Doc]:
13    """Stage 1 — read raw shards from disk, one document per line."""
14    for p in paths:
15        with open(p, encoding="utf-8", errors="ignore") as f:
16            for i, line in enumerate(f):
17                yield f"{p}:{i}", line.strip()
18
19def language_filter(docs: Iterator[Doc]) -> Iterator[Doc]:
20    """Stage 2 — crude language guess: keep ASCII-heavy English."""
21    for did, text in docs:
22        if not text:
23            continue
24        ascii_frac = sum(c.isascii() for c in text) / max(1, len(text))
25        if ascii_frac > 0.9:
26            yield did, text
27
28def heuristic_filter(docs: Iterator[Doc]) -> Iterator[Doc]:
29    """Stage 3 — drop too-short or highly repetitive docs (SEO spam)."""
30    for did, text in docs:
31        if len(text) < 200:
32            continue
33        words = text.split()
34        uniq = len(set(words)) / max(1, len(words))
35        if uniq < 0.30:
36            continue
37        yield did, text
38
39def exact_dedup(docs: Iterator[Doc]) -> Iterator[Doc]:
40    """Stage 4 — drop bit-identical documents seen earlier in the stream."""
41    seen: set[str] = set()
42    for did, text in docs:
43        h = hashlib.sha1(text.encode()).hexdigest()
44        if h in seen:
45            continue
46        seen.add(h)
47        yield did, text
48
49def tokenize_count(docs: Iterator[Doc]) -> Iterator[tuple[str, int]]:
50    """Stage 5 — stand-in for a real BPE tokenizer."""
51    for did, text in docs:
52        n = len(re.findall(r"\S+", text))   # whitespace token count
53        yield did, n
54
55def pipeline(paths: list[str]) -> Iterator[tuple[str, int]]:
56    return tokenize_count(
57        exact_dedup(
58            heuristic_filter(
59                language_filter(
60                    ingest(paths)))))
61
62if __name__ == "__main__":
63    paths = ["data/crawl_shard_001.txt"]
64    docs_kept, tokens_kept = 0, 0
65    for did, n in pipeline(paths):
66        docs_kept   += 1
67        tokens_kept += n
68    print(f"docs kept   = {docs_kept}")
69    print(f"tokens kept = {tokens_kept}")

Three architectural points are worth marking. First, nothing in this pipeline materializes a list. Every stage is lazy. That is what makes it scale: peak memory is a constant function of the per-document size, not the corpus size. Second, the order of stages matters economically. Language ID and exact dedup both throw away half their input but cost <1 µs per doc; a perplexity-based quality model costs ~1 ms per doc. Putting the cheap filters first amortizes the expensive ones, which is why every production pipeline from Dolma to NeMo Curator runs the cheap stages early. Third, the "DAG" is just function composition. Production frameworks add scheduling, retries, sharded shuffles, and provenance tracking, but every one of them ultimately compiles down to a chain of operators identical in shape to the chain above.

Where this code grows up. The same five-function skeleton — ingest, filter, dedup, score, tokenize — appears in every published pretraining pipeline. RedPajama-Data, Dolma, FineWeb, Nemotron Curator, and DeepSeek-V3's internal toolchain are all variations on this theme with industrial scheduling on top.

PyTorch: Sharded IterableDataset and the Train Loader

Once the offline pipeline has produced tokenized shards on disk, the online half takes over. Its only job is to stream those shards into the GPU at line rate, without ever replaying tokens and without ever stalling. PyTorch's IterableDataset + DataLoader is the canonical implementation. The complexity is not in the iterator body — it is in the sharding logic that ensures each token reaches exactly one GPU per epoch.

1import numpy as np
2import random
3import torch
4import torch.distributed as dist
5from torch.utils.data import IterableDataset, DataLoader
6
7class PackedShardDataset(IterableDataset):
8    """
9    Stream pre-tokenized .bin shards (uint32 arrays of token IDs) and pack
10    them into fixed-length sequences. Each (rank, worker) sees a disjoint
11    slice of shards, so the model sees every token exactly once per epoch.
12    """
13    def __init__(self, shard_paths: list[str], seq_len: int, seed: int):
14        super().__init__()
15        self.shard_paths = sorted(shard_paths)
16        self.seq_len     = seq_len
17        self.seed        = seed
18
19    def _my_shards(self) -> list[str]:
20        rank   = dist.get_rank()        if dist.is_initialized() else 0
21        world  = dist.get_world_size()  if dist.is_initialized() else 1
22        info   = torch.utils.data.get_worker_info()
23        wid    = info.id           if info is not None else 0
24        nw     = info.num_workers  if info is not None else 1
25        # Two-level partition: across DDP ranks, then across DataLoader workers.
26        mine = [
27            p for i, p in enumerate(self.shard_paths)
28            if i % world == rank and (i // world) % nw == wid
29        ]
30        random.Random(self.seed + rank * 1009 + wid).shuffle(mine)
31        return mine
32
33    def __iter__(self):
34        buf: list[int] = []
35        for shard in self._my_shards():
36            arr = np.memmap(shard, dtype=np.uint32, mode="r")
37            buf.extend(arr.tolist())
38            while len(buf) >= self.seq_len + 1:
39                window = buf[: self.seq_len + 1]
40                buf    = buf[self.seq_len :]                # advance by seq_len
41                x = torch.tensor(window[:-1], dtype=torch.long)
42                y = torch.tensor(window[1:],  dtype=torch.long)
43                yield x, y
44
45ds = PackedShardDataset(
46    shard_paths=[f"/data/shards/shard_{i:05d}.bin" for i in range(8192)],
47    seq_len=4096,
48    seed=20250517,
49)
50loader = DataLoader(
51    ds,
52    batch_size=2,             # per-GPU micro-batch
53    num_workers=4,            # CPU workers per GPU
54    pin_memory=True,
55    persistent_workers=True,
56    prefetch_factor=4,
57)
58for step, (x, y) in enumerate(loader):
59    # x.shape == y.shape == (B, T) — packed, sharded, shuffled. Train as usual.
60    pass

Two subtleties recur in every real loader and are worth surfacing:

1. Token-level shuffling is wrong. If you naïvely shuffle across token boundaries, you destroy long-range structure within documents — the very signal a 4096-context transformer is supposed to model. The shuffle granularity is the shard, sometimes the document, never the token. The packed-sequence loader above preserves the original token order within each document and only shuffles the order in which documents are streamed.
2. Resumability is part of the loader, not the trainer. When training crashes (and at 54 days of wall-clock, it will), you must resume from the last token boundary, not the last batch. Production loaders persist a (shard_index, byte_offset) pair per (rank, worker) into the checkpoint and reseed exactly there on restart. Get this wrong and you replay or skip millions of tokens silently, which surfaces as mysterious loss spikes weeks later.

What Changes at Massive Scale

At toy scale the pipeline is a Python script that runs in twenty minutes. At 14.8T-token scale it is its own distributed system, with operational constraints that look more like a database than a script. Four things change as the corpus grows.

The offline stage becomes a job scheduler

Past about 100 TB, no single host can store the working set. Stages become Spark / Ray / Beam jobs with explicit input and output shard manifests; intermediate results are written to object storage and consumed by the next stage. Provenance tracking — for each surviving token, which raw URL did it come from? — becomes a hard requirement for contamination detection (Section 8.7). The pipeline stops being a program and starts being a workflow.

Dedup becomes the bottleneck

Exact dedup over 6 PB needs roughly $6 \times 10^{11}$ document hashes in a single set. That does not fit in memory on any host. The fix is sorted external merge or a sharded RocksDB / Bloom-filter approach. Near-dedup is worse: MinHash LSH over a billion documents requires a permutation-based signature for each doc and a multi-pass band-collision search. Section 8.2 spends its entire length on this stage because, at frontier scale, dedup typically costs more CPU than every other stage combined.

The online loader becomes I/O-bound, not CPU-bound

At 2048 GPUs the aggregate read is ~12 GB/s. A single NVMe SSD does ~3 GB/s; a single S3 bucket region throttles around 25 GB/s. The loader's architecture splits: hot shards live on per-host NVMe (read locally by the GPU's own workers), cold shards live in object storage and are prefetched ~10 minutes ahead of consumption. Some teams add a CDN-style cache tier in between. The DataLoader code looks unchanged; the storage layer underneath becomes a multi-tier system whose design is the difference between 95% and 99% MFU.

Failure isolation becomes a design goal

At 2048 GPUs, you will lose roughly one GPU per day to ECC errors, cable flaps, or NCCL deadlocks. The loader must not lose its place when its host dies. The fix is deterministic per-shard ordering plus a step-counter checkpoint: any restarted rank can reconstruct exactly which token it was about to read by replaying its sharding function with the same seed. The Python code in the previous section already bakes this in (seed + rank*1009 + wid); the engineering reality is that without it, every node failure costs hours of re-shuffle and risks replaying tokens.

Engineering Reality: Where Pipelines Actually Break

Three failure modes account for the majority of real-world pipeline incidents at frontier scale. Each of them is invisible in the math and only becomes obvious in the engineering.

1. Silent token replay. A bug in the rank-/worker-mod sharding sends the same shard to two workers. The model still trains, the loss still drops, but it drops on a corpus that is effectively 1.x epochs of the same data — equivalent to training on 30–50% less unique text than the spec promised. The only diagnostic is a per-shard usage counter compared against an expectation; the fix is unit tests on the sharding function and reproducibility checks across restarts.
2. Quality drift across the run. If the offline pipeline produces shards in the order they were processed (cheap first, expensive later), then shard $i$ has systematically different quality than shard $i + k$. Without a global shuffle, the model sees easy text early and hard text late, which interacts badly with the learning-rate schedule (Chapter 11). The fix is a corpus-wide shuffle done once at the end of the offline pipeline, before shards are written to their final paths.
3. Loader stalls disguised as training stalls. When the loader cannot keep up — usually because cold shards have not been prefetched in time — GPUs idle for a few hundred milliseconds per step. NCCL allreduces still complete, the train loop still ticks, but MFU drops from 40% to 30%. The diagnosis is to add per-step loader latency telemetry; the fix is to widen the prefetch window or warm the storage cache. This is the most common reason a model team mistakes a data problem for an optimizer problem.

One sentence to carry forward into the rest of the chapter: the pipeline is a contract between the offline retention math and the online throughput math, and every subsequent design choice — dedup algorithm, quality classifier, mixing weights, curriculum, contamination check — is a renegotiation of one side of that contract. The remaining sections of Chapter 8 walk through each renegotiation in order.