Large-scale system optimization
Every idea in the previous eighteen lessons quietly assumed it would scale for free. It does not. At a billion items and a hundred thousand queries per second, the model becomes a system — and four numbers (cardinality, dimension, recall, dollars) decide the whole design.
Scale turns ideas into arithmetic
Scale does not invent new ideas. It converts the ideas you already have into arithmetic problems, and then makes one term in the arithmetic too big to ignore. There are only three lines you ever need to write:
Once those three lines exist for your system, every design decision — shard or replicate, ANN or brute force, quantize or distill, cache or recompute — is just the answer to a single question: which term is too big, and which lever shrinks it? This lesson walks the funnel from 04 · Ranking models and the index from 03 · Embeddings & ANN one rung deeper — into the numbers that bound the fleet.
The cascade is a compute budget — and why 粗排 exists
01 · The funnel introduces the four stages conceptually; this lesson owns the engineering of the one in the middle that beginners skip. The industrial pipeline is not retrieval → ranker. It is a four-stage cascade, and the whole point is that each stage spends compute inversely to how many candidates it sees: a cheap model on many items, an expensive model on few.
The arithmetic forces the shape. Suppose the heavy fine-ranker (精排) costs ≈ 2 MFLOPs/candidate (cross layers, attention towers, hundreds of features). Running it on the 10³ survivors of retrieval is fine: 10³ × 2×10⁶ = 2×10⁹ FLOPs/query. Running it on all 10⁸ retrieved items would be 10⁸ × 2×10⁶ = 2×10¹⁴ FLOPs/query — a 100,000× blow-up that no SLO survives. So a stage must sit between retrieval and the heavy ranker to cut 10³ → 10² cheaply. That stage is the coarse-ranker (粗排).
Twin-tower with distillation — the coarse-ranker's design
The coarse-ranker's budget (~10–20 ms on 10³ items) rules out the heavy ranker's trick of cross features — features that mix user and item (e.g. user × item_category), which must be recomputed for every (user, item) pair. With 10³ pairs that is 10³ fresh feature-crosses + forward passes per query: too slow. So the coarse-ranker is a twin-tower (双塔) net — a user tower and an item tower that never interact until a final dot-product:
score(u, i) = e_user(u) · e_item(i)
This is what makes it cheap. The item tower is query-independent: every item's vector e_item(i) is precomputed offline and cached. At request time you run the user tower once, then score 10³ items as 10³ dot-products — 10³ × d multiply-adds, not 10³ forward passes. At d = 64 that is 10³ × 64 ≈ 6.4×10⁴ FLOPs/query for the scoring, versus the heavy ranker's 2×10⁹. Cheap enough to score everything retrieval hands you.
The accuracy cost of dropping cross features is recovered by distillation (the same teacher→student trick used later for serving): the heavy fine-ranker is the teacher, the twin-tower coarse-ranker is the student, and the student is trained to reproduce the teacher's scores or ranking order rather than just the raw click label. The coarse-ranker thus learns "what the expensive model would have said" within the limits of a dot-product — far better than training it on clicks alone.
Retrieval itself is 02 · Candidate generation (the multi-route 召回 that produces the 10³), and its dominant cost — the ANN dot-product — is the next several sections. The 重排 (rerank) stage is business rules and diversity, covered in 20 · Fairness & diversity. From here on, "ranker" means the heavy 精排 model unless stated.
The embedding table is the elephant
From 04 · Ranking models, the headline fact: a modern ranker is an enormous sparse embedding store with a small dense network on top. Every high-cardinality ID — user, item, ad, author, query token — is a learned row in a table. The dense MLP is megabytes. The embedding table is hundreds of gigabytes to terabytes. Its size is pure multiplication:
table_bytes = cardinality × dim × bytes_per_value
A platform with 1 billion users and 1 billion items, each embedded as a 128-dim fp32 vector:
2 × 10⁹ × 128 × 4 bytes = 1.024 × 10¹² ≈ 1 TB
One terabyte of parameters, of which the actual network might be 200 MB — a factor of five thousand. This single imbalance reshapes everything downstream: the table cannot live where the model lives.
Sharding the table
Embedding lookups are independent — a request for item 847,221,003 touches exactly one row — so the table shards cleanly by ID. Row i lives on machine i mod S. A forward pass gathers each example's IDs, routes each lookup to the shard that owns it, pulls the vectors back, and feeds them to the dense model; gradients scatter back the same way. This is embedding-table (model) parallelism — the reason large recommenders are a distributed key-value store of parameters, not a single checkpoint.
The cost it creates is communication. Every step, every worker fetches vectors from every shard (the gather / all-to-all) and pushes gradients back (the scatter). That all-to-all traffic — not the matmuls — is the bottleneck of training a recommender at scale, and it forces the next decision.
Shrinking the table itself — hashing trick vs full embedding
Sharding makes the 1 TB table fit, but it does not make it smaller — you still pay for every row. The lever that shrinks the row count is how you map a billion IDs to embedding rows in the first place. Two options sit at opposite ends of a memory/accuracy trade-off:
| Full embedding table | Hashing trick | |
|---|---|---|
| Mapping | Each ID owns a private row (a stored id→row map) | row = hash(id) mod B — no stored map |
| Memory | cardinality × dim × bytes — grows with every new ID | B × dim × bytes — fixed, independent of cardinality |
| Accuracy | Best — each ID is independently learnable | Lossy — IDs colliding into one bucket share a row, blurring their distinction |
| Cold start | New ID has no row → needs special handling | New ID hashes to an existing bucket → degrades gracefully, can't be fine-tuned alone |
The arithmetic is the whole argument. A full table for 10⁹ item IDs at 64-dim fp32 is 10⁹ × 64 × 4 = 256 GB. Hash the same IDs into B = 10⁷ buckets and the table is 10⁷ × 64 × 4 = 2.56 GB — a 100× shrink, flat no matter how many new IDs arrive. The bill is collisions: with 10⁹ IDs in 10⁷ buckets, roughly 10⁹ / 10⁷ = 100 IDs share each row on average, so a hashed bucket is a blurred average of ~100 items that the model can no longer tell apart.
Two ways to do distributed training
The dense model is small and identical everywhere; the embedding table is huge and split. Those two facts pull toward two different distributed-training patterns, and large recommenders use both at once.
| Parameter server (async) | All-reduce (sync) | |
|---|---|---|
| Picture | Workers push / pull params to central server shards | Workers exchange gradients peer-to-peer in a ring |
| Best for | The huge sparse embedding table (TB, sharded by ID) | The small dense network (replicated on every worker) |
| Comms cost | Each worker touches only the shards holding its IDs — sparse, cheap | Every gradient summed across all workers — ≈ 2(W−1)/W · model_bytes per step |
| Consistency | Stale / async updates — tolerable for rows touched rarely | Exact, synchronous — required for shared dense weights |
| Failure mode | Server hot-spotting on popular IDs; staleness | One slow worker (straggler) stalls everyone |
The reason the split works: replicating a 1 TB table on every worker is impossible, but all-reduce on a 200 MB dense model is cheap. Conversely, routing every dense gradient through a central server makes the server a bottleneck, but the sparse table — where each example touches only a handful of a billion rows — is exactly what a parameter server was built for. Sparse → parameter server; dense → all-reduce. That hybrid is the standard industrial layout (TorchRec / HugeCTR for the tables, ring all-reduce for the MLP).
The 2(W−1)/W factor is worth internalizing: ring all-reduce sends each shard of the gradient around the ring once to reduce it and once to broadcast the sum, so per-worker traffic is ≈ 2 · model_bytes regardless of how many workers W you add. That is why all-reduce scales — comms are flat in W — but only for a model small enough to replicate.
ANN at scale — because O(N·d) is a wall
Retrieval (the front of the funnel from 03 · Embeddings & ANN) must turn a billion-item corpus into ~1,000 candidates in a few milliseconds. The honest way — dot-product the user vector against every item — is brute force, and its cost is unforgiving:
brute force = N × d multiply-adds = 10⁹ × 128 ≈ 1.3 × 10¹¹ FLOPs / query
At 100k QPS that is 1.3 × 10¹⁶ FLOP/s for retrieval alone — a wall of GPUs to answer the cheapest stage of the funnel. The escape is to give up exactness: Approximate Nearest Neighbor (ANN) returns almost the true top-K, trading a few percent of recall for two-to-three orders of magnitude of speed. Three families dominate, and they sit at different corners of the same triangle.
| Method | Idea | Query cost | Recall | Memory |
|---|---|---|---|---|
| IVF (inverted file) | K-means the corpus into ~√N cells; scan only the nprobe cells nearest the query | O(√N · d) | tunable via nprobe | full vectors (high) |
| PQ (product quantization) | Split each vector into m sub-vectors; replace each with a 1-byte codebook index | cheap table-lookup distances | lossy (compression noise) | tiny (see below) |
| HNSW (graph) | A navigable small-world graph; greedily hop toward the query through layers | O(log N · d) | very high | vectors + graph edges (highest) |
In practice they compose: IVF-PQ (cells to prune the search space, PQ to shrink the vectors) is the standard billion-scale recipe in FAISS; HNSW is the latency champion when RAM is plentiful and is what GPU-accelerated libraries push under 10 ms on a billion items.
Product quantization — the compression that makes it fit
Even with IVF pruning the search, you still have to store a billion vectors. A billion 128-dim fp32 vectors is 10⁹ × 128 × 4 = 512 GB — too big for one machine's RAM, let alone a GPU. PQ shrinks it. Split each 128-dim vector into m = 16 sub-vectors of 8 dims; cluster each sub-space into 256 centroids; store only the 1-byte index of the nearest centroid per sub-vector:
512 bytes/vec → 16 bytes/vec ⇒ 32× compression | 512 GB → 16 GB (now fits on one GPU)
The compression ratio is (d × 4) / m bytes — choose m and the codebook size to slide between memory and recall. Distances are computed against the codebook by table lookup, so PQ is fast and small. The price is reconstruction error — you compare against the centroid, not the original vector — which shaves a few points of recall. At a billion items, that deal is almost always worth taking.
ItemCF at scale — the O(N²) similarity matrix
Not every retrieval route is a learned embedding. A workhorse recall source is item-to-item collaborative filtering (ItemCF): "users who interacted with this item also interacted with…", scored by a precomputed item–item similarity. It is a different memory trade-off from the embedding table — here the elephant is a matrix, and its cost is quadratic in the catalog.
The full similarity matrix is O(N²): every item against every other item. At N = 10⁷ items that is
N² = (10⁷)² = 10¹⁴ entries — at 4 bytes each, 4×10¹⁴ B = 400 TB
Infeasible to store, and infeasible to compute densely. But the matrix is mostly worthless: an item is meaningfully similar to only a few hundred others, and the rest of the row is near-zero noise. That observation is the whole optimization. Four moves turn 400 TB into something you can serve:
| Move | Mechanism | Effect |
|---|---|---|
| Block by category | Only compute similarity within a category / popularity bucket, not across the whole catalog | Splits one N² into many small n²; cross-block pairs (almost all of them) are never formed |
| Top-K truncation | Store only each item's K most-similar neighbours (K = 100–500), discard the rest of the row | N×N → N×K — see arithmetic below |
| Incremental delta updates | Recompute only rows touched by new behaviour (co-occurrence deltas), not the whole matrix nightly | Update cost scales with activity, not with N² |
| Serve via ANN | Treat each item's neighbour list as vectors; online "find similar items" becomes a FAISS lookup | Reuses the ANN machinery above — no bespoke serving path |
The top-K truncation is the load-bearing step. Keep K = 200 neighbours per item instead of the full row of N:
N × K × (4 + 4) bytes = 10⁷ × 200 × 8 = 1.6×10¹⁰ B ≈ 16 GB
(4 bytes for the neighbour's item-id, 4 for the similarity score.) 400 TB → 16 GB, a ~25,000× shrink, and it now fits in RAM and serves as a vector lookup. The cost is exactly what truncation always costs: you lose the long-tail similarities below rank K — but those were the near-zero noise you wanted gone, so for ItemCF the deal is almost free. The pattern generalizes to any precomputed co-occurrence matrix: block, truncate to top-K, update by deltas, serve through the ANN index.
Graph importance-pruning for recall
A graph recall route (user–item bipartite graph, propagate to find candidates — see 24 · GNN) has the same disease in edge form: a dense graph has too many edges to propagate over in the latency budget. The fix mirrors top-K truncation but on edges: score each edge's importance — e.g. with Random-Walk-with-Restart (RWR) visit probability — and prune low-weight edges before propagation, keeping only edges above a threshold (or each node's top-K edges). Production graph-recall systems report pruning on the order of 95% of edges while preserving the high-importance paths that actually carry signal — turning an intractable propagation into a cheap one, the same precompute-and-truncate trade-off applied to graph structure rather than a matrix.
Interactive · ANN recall vs latency vs RAM
Watch the per-query cost — and the GPU count needed to hold the QPS — collapse as you move from an exact scan to a graph index. Then watch recall and memory pay for it. The numbers are order-of-magnitude, not literal.
Compressing the ranking model for serving
Retrieval solved, the ranker (from 04 · Ranking models) now scores ~1,000 candidates per request under the latency SLO. A heavy DLRM / DCN-v2 with cross layers and attention towers is too slow to run a thousand times per query at 100k QPS. Three compression levers, each with a different accuracy bill:
| Technique | What it does | Win | Accuracy cost |
|---|---|---|---|
| Quantization | fp32 → fp16 → int8 weights & activations | 2–4× smaller & faster; int8 uses fast integer units | tiny (fp16 ~free; int8 needs calibration) |
| Distillation | Train a small student to mimic a big teacher's soft scores | 10–100× fewer params, much faster | moderate — student trails teacher by a small gap |
| Pruning | Zero out low-importance weights / experts / heads | sparse, smaller; can drop whole sub-towers | low if gradual; needs sparse kernels to pay off |
They stack. A typical serving model is distilled from the offline teacher, then quantized to int8, then deployed with operator fusion (TensorRT). A reported real-world combination — int8 quantization plus a higher cache-hit rate — dropped ranking latency from ~120 ms to ~65 ms with no measurable AUC loss, which then lifts the business metric purely by serving more requests, faster.
Caching — the cheapest FLOP is the one you skip
Recommendation traffic is brutally skewed: a tiny fraction of users and items generate most requests. That skew is a gift — a small cache absorbs a large share of the work. Three layers worth caching:
- User-feature embeddings — a returning user's profile vector changes slowly; cache it (Redis + node-local) and skip the feature fetch.
- Item / ANN results — the candidate set for a popular query or trending item is the same for thousands of requests; cache the retrieval output.
- Full responses — for logged-out or near-duplicate contexts, cache the final ranked list with a short TTL.
The arithmetic is Amdahl's law on cost: with hit-rate h and a near-free hit, effective compute per query is (1 − h) of the uncached cost. A 70% hit-rate on the heaviest stage cuts its bill by 70%. The tension: caching trades freshness for cost — too long a TTL and you serve stale recommendations into a fast-moving feed. The TTL is that dial.
The $/query bound that closes the loop
Everything above serves one number the business actually feels: cost per query. It is the funnel turned into a bill — sum the compute of each stage, divide by what a GPU-second buys:
$/query = Σ_stage (FLOPs_stage / achieved_FLOP/s) × $/GPU-sec
GPUs = QPS × (compute/query) / (FLOP/s/GPU × utilization)
This is why every optimization in this lesson exists, and each attacks a specific term in that sum:
| Lever | Term it shrinks | Typical factor |
|---|---|---|
| Sharding | makes the table fit at all (memory) | system exists / doesn't |
| ANN vs brute force | retrieval's N·d | ~100–1000× |
| PQ compression | index memory → fewer machines | ~32× |
| Distillation + int8 | ranking's compute / candidate | ~10–100× × 2–4× |
| Caching | every stage, by hit-rate | (1 − h) |
Interview prompts you should be ready for
- "Your model is 1 TB. Where does that come from, and how do you train it on a cluster of 80 GB GPUs?" (table_bytes = cardinality × dim × bytes; it's the embedding table, not the MLP. Shard the table by ID across machines — embedding model-parallelism — all-to-all the looked-up rows, and data-parallel the small dense top with all-reduce.)
- "Why a parameter server for the embeddings but all-reduce for the dense net?" (Sparse table is too big to replicate and each example touches only a few rows → PS with cheap sharded access, staleness tolerable. Dense net is small, shared, needs exact sync → all-reduce, comms ≈ 2(W−1)/W · model_bytes, flat in W.)
- "Estimate the comms per step for ring all-reduce on a 200 MB dense model across 256 workers." (≈ 2 · (255/256) · 200 MB ≈ 398 MB per worker per step; at 50 GB/s that's ~8 ms. The point of the 2(W−1)/W factor: traffic is independent of W, which is why all-reduce scales.)
- "Brute-force retrieval on 10⁹ items at dim 128 — why can't you ship it, and what do you ship instead?" (N·d ≈ 1.3×10¹¹ FLOPs/query; × QPS = a GPU wall just for retrieval. Ship ANN — IVF-PQ or HNSW — trading a few % recall for 100–1000× speed.)
- "Walk me through the recall–latency–memory triangle. Which method for which constraint?" (HNSW: best recall+latency, costs RAM for the graph. PQ: best memory+latency, costs recall via lossy codebook. IVF nprobe: directly dials recall vs latency. Pick the corner your binding constraint allows; IVF-PQ composes two of them.)
- "How does PQ get 32× compression, and what does it cost?" (Split 128-dim vector into m=16 sub-vectors, store the 1-byte index of the nearest of 256 centroids each: 512 → 16 bytes. Cost is reconstruction error — you compare against centroids — losing a few points of recall.)
- "Ranking is too slow at 1,000 candidates × 100k QPS. Three things you'd try, with the accuracy cost of each?" (Quantize fp32→int8: 2–4× faster, tiny cost with calibration. Distill teacher→student: 10–100× fewer params, small gap. Prune weights/experts: smaller, low cost if gradual but needs sparse kernels. They stack: distill, then int8, then fuse.)
- "You add a 60% cache hit-rate to the heaviest stage. What happens to its cost, and what's the risk?" ((1 − h) = 40% of the original bill on that stage. Risk is staleness — TTL trades freshness for cost; too long and you serve stale lists into a fast-moving feed.)
- "Why is there a coarse-ranker (粗排) between retrieval and the heavy ranker — couldn't you just run the good model on all retrieved candidates?" (Heavy ranker is ~2 MFLOPs/candidate, built for ~10²; on 10³ retrieved items that's a 100,000× blow-up vs running it on the survivors, and merged multi-route retrieval scores are on incomparable scales so you can't trust them for the top-100 cut. The coarse-ranker is a cheap twin-tower — item vectors precomputed, score = one user-tower pass + 10³ dot-products, no cross features — distilled from the fine-ranker so it learns "what the expensive model would say." Failure mode: coarse/fine inconsistency — it drops items the fine ranker would have loved, capping quality; fix by training on the fine-ranker's order and monitoring top-K overlap.)
- "ItemCF at 10⁷ items — the similarity matrix is 400 TB. How do you serve it?" (O(N²) is mostly noise; an item is similar to only a few hundred others. Block by category so cross-block pairs never form; truncate to top-K=100–500 neighbours per item → N×K not N×N → ~16 GB at K=200; update incrementally from behaviour deltas not full nightly recompute; serve the neighbour lists through FAISS so it reuses the ANN path. Cost is losing sub-rank-K similarities — exactly the noise you wanted gone.)
- "The embedding table is 256 GB and growing. Beyond sharding, how do you cap its memory?" (Sharding makes it fit, not smaller. Hashing trick: row = hash(id) mod B, fixed B×dim×bytes regardless of cardinality — 100× shrink at B=10⁷ — but ~100 IDs/bucket collide into a blurred shared row, no per-id fine-tuning. So go hybrid: head/frequent IDs get private rows (most traffic, enough data to learn), tail/rare IDs share hashed buckets (noisy anyway, rarely seen). A parameter server promotes a tail ID to a private row once it crosses a frequency threshold.)