Problem Definition

"You scrolled, and I had thirty milliseconds. In that blink I asked a hundred machines what you might like, watched a billion items shrink to twenty, and handed you a row of suggestions as if I had been thinking about you all along."

A Recommendation, Assembled in Eighty Milliseconds From a Hundred Machines

A recommender system answers a question that has no query text: given who you are and what you are looking at right now, which few items, out of a catalog of billions, should fill this row of the screen. On a laptop, over a few thousand products and a few hundred users, this is a weekend project: learn a small matrix factorization, score every item for every user, and sort. The interesting version, and the one this chapter studies, is the version that powers a large marketplace, video platform, or social feed. When the catalog is measured in billions of items, the audience in hundreds of millions of users, and the response budget in tens of milliseconds, every component that was trivial on the laptop becomes a distributed system in its own right. The purpose of this case study is to show how the machinery built in Parts III through VII assembles into one coherent product, and where the seams between those parts actually fall.

We will treat the system as a concrete target rather than an abstraction. Throughout the chapter, the running specification is a recommendation service with a catalog on the order of two billion items, an active user population in the hundreds of millions, a serving path that must return a ranked slate within a latency a person never notices, a peak throughput measured in hundreds of thousands of requests per second, and a freshness requirement that lets a click recorded seconds ago shape the next recommendation. Those five pressures, catalog and user scale, latency, throughput, freshness of signals, and cost, are the requirements; the rest of the chapter is the design that satisfies them.

1. The Requirements: Five Numbers That Rule Out One Machine Beginner

A problem definition is only useful if it commits to numbers, because the numbers decide which ceilings bind and therefore which axes of distribution we must reach for. We fix five requirements and carry them through the entire chapter. The first is scale: the system learns and serves embeddings for a catalog of $N_{\text{items}} \approx 2 \times 10^{9}$ items and $N_{\text{users}} \approx 4 \times 10^{8}$ users, plus categorical features (the advertiser, the query, the device, the context) whose vocabularies add billions more rows. The second is latency: the end-to-end serving budget, from request to ranked slate, sits on the order of tens of milliseconds, because a recommendation that arrives after the page has rendered is a recommendation nobody saw. The third is throughput: the service must sustain hundreds of thousands of requests per second at peak, every one of which fans out across the retrieval and ranking fleet. The fourth is freshness: a user who just clicked, skipped, or purchased should see that signal reflected within seconds, so the system is never a nightly batch but a continuously updated flow. The fifth is cost: the design must hit the first four at a price an operator would pay, which immediately disqualifies any approach that scores the whole catalog per request or holds a private copy of the embedding tables on every server.

These five requirements interact, and the interaction is the whole subject. Freshness fights cost, because applying every click to the embedding tables in real time burns memory bandwidth and network traffic. Latency fights catalog size, because a larger catalog means more candidates to consider per request. Throughput fights latency, because every concurrent request competes for the same retrieval and ranking hardware. A single machine fails all five at once and for independent reasons: it cannot store the embedding tables, cannot scan billions of candidates inside the budget, cannot apply the update stream, and cannot serve the request volume. The next section makes that failure quantitative.

2. Why a Single Machine Is Not Even Close Beginner

The argument for distribution here is not rhetorical; it falls out of arithmetic on the five requirements. Take the embedding tables first. Every id the model knows about, an item, a user, an advertiser, a query token, gets a learned vector of dimension $d$ stored at $b$ bytes per dimension. The number of rows across all tables and the table size in bytes are

where $N_{\text{feat}}$ collects the categorical-feature vocabularies. With billions of rows, a dimension $d = 256$, and $b = 4$ bytes per dimension in single precision, $S_{\text{emb}}$ runs to several terabytes, one to two orders of magnitude past the memory of any single node. The number of parameter-server shards the table must be split into is the table size divided by the memory each node can devote to it,

which for a node offering tens of gigabytes of fast memory to embeddings lands in the dozens to low hundreds. The serving path is no kinder. It cannot score all $N_{\text{items}}$ candidates per request, so it runs a funnel that shrinks the catalog through successive stages, retrieval then ranking then final selection, with a multiplicative reduction

that takes billions of items down to the few tens actually shown. Even the surviving candidates obey a concurrency wall. The query path follows Little's law: the number of requests in flight at any instant is the arrival rate times how long each one stays in the system,

so a throughput $\lambda$ of hundreds of thousands of requests per second at a latency $L$ of tens of milliseconds keeps thousands of requests being served concurrently, which no single server's compute or memory can hold. Figure 38.1.1 shows the multi-stage pipeline and the axis each stage loads; Code 38.1.1 puts real numbers behind every one of these formulas.

The diagram fixes the vocabulary for the chapter. The terabyte-scale embedding tables at the top are the model that cannot live on one machine and that every other stage reads from. The middle row is the per-request serving funnel that runs under the latency SLO. The lower band is the streaming feature and freshness layer that keeps the model current; the bottom strip is the online evaluation that keeps the whole thing honest. The numbers on each box come from the same back-of-envelope model, which we now run with real arithmetic so that no estimate in this chapter is a guess.

import math

# --- Catalog and user scale (the five requirements) ---
N_items   = 2_000_000_000    # catalog size (billions of items)
N_users   = 400_000_000      # active users (hundreds of millions)
N_feat    = 4_000_000_000    # extra categorical-feature ids (ad, query, context)
dim       = 256              # embedding dimension d
bytes_id  = 4                # fp32 per dimension, b
qps       = 300_000          # peak queries per second, lambda
latency   = 0.030            # end-to-end SLO budget, seconds (tens of ms)
mem_node  = 64               # GB of fast memory a node devotes to embeddings

# --- Embedding table size: rows R = items + users + features ---
R       = N_items + N_users + N_feat
S_emb   = R * dim * bytes_id                    # table size in bytes
emb_tb  = S_emb / 1e12
shards  = math.ceil(S_emb / (mem_node * 1e9))   # K_shard parameter-server shards

# --- Candidate funnel: catalog -> retrieved -> ranked -> shown ---
funnel  = [N_items, 10_000, 500, 20]
rho     = [funnel[j] / funnel[j + 1] for j in range(len(funnel) - 1)]

# --- Concurrency from Little's law: C = lambda * L ---
C = qps * latency

# --- Cost per 1000 requests at an assumed serving-fleet rate ---
fleet_nodes   = 1200
node_per_hour = 3.50                              # USD per node-hour
cost_per_1k   = (fleet_nodes * node_per_hour) / (qps * 3600) * 1000

print(f"embedding rows R (all tables): {R:,}")
print(f"embedding table S_emb        : {emb_tb:,.1f} TB")
print(f"parameter-server shards      : {shards:,}")
print(f"candidate funnel             : {' -> '.join(f'{x:,}' for x in funnel)}")
print(f"per-stage reduction factors  : {' , '.join(f'{r:,.0f}x' for r in rho)}")
print(f"queries in flight C=QPS*L    : {C:,.0f}")
print(f"cost per 1000 requests       : ${cost_per_1k:.3f}")

embedding rows R (all tables): 6,400,000,000
embedding table S_emb        : 6.6 TB
parameter-server shards      : 103
candidate funnel             : 2,000,000,000 -> 10,000 -> 500 -> 20
per-stage reduction factors  : 200,000x , 20x , 25x
queries in flight C=QPS*L    : 9,000
cost per 1000 requests       : $0.004

The output settles the question. The embedding tables are $6.6$ terabytes, which alone exceeds the memory of any single accelerator by one to two orders of magnitude and forces the table across roughly one hundred shards before training or serving can begin. The serving funnel must reduce two billion candidates to twenty, a total compression of a hundred million to one, with the first and largest cut, a factor of two hundred thousand, taken at retrieval rather than by brute-force scoring. The query path keeps thousands of requests in flight simultaneously. None of these is a marginal overshoot that a faster chip would absorb; each is one to several orders of magnitude beyond a single node, and each lands on a different stage of the pipeline. That separation is exactly what makes the six-axis decomposition the right tool.

3. Decomposing the Pipeline Onto the Six Axes Intermediate

The six axes of distribution from Section 1.1, distribute data, distribute training, distribute the model, distribute inference, coordinate the cluster, and distribute intelligence, give us a map onto which every stage of Figure 38.1.1 places cleanly. Mapping the stages is the planning act of the whole chapter: it tells us which earlier part of the book owns each stage and which later section of this chapter develops it. Table 38.1.1 is that map, and it doubles as the chapter's table of contents.

Reading the table top to bottom traces the dataflow of Figure 38.1.1, and reading the third column traces the path back through the book. Learning the embeddings is the act of distributing a model too large for one node: the tables are sharded across parameter servers that workers push gradients to and pull current vectors from, the push-pull pattern of Chapter 11, here scaled to billions of rows. Candidate generation is approximate nearest-neighbor retrieval over a sharded vector index, the scatter-gather across shards that Chapter 25 owns, and its tight latency makes it primarily a cluster-coordination problem. Ranking is distributed inference: a deep scoring model replicated across the fleet and run on the few hundred candidates the funnel let through, drawing on the inference-systems and serving economics of Chapter 23 and Chapter 24. The feature store is distributed data, the storage and loading machinery of Chapter 8 specialized to serve features at both training and request time. Real-time personalization is stream processing, the online-AI material of Chapter 9 that turns a click seconds old into a fresh feature. Evaluation closes the loop with the methodology of Chapter 5. No stage is new machinery; the contribution of this chapter is the assembly and the seams.

4. The Training Path Versus the Serving Path Intermediate

One structural decision organizes the rest of the chapter, and it is visible in Figure 38.1.1 as the difference between the streaming layer that updates the tables and the serving funnel that reads them. The training path and the serving path have opposite cost profiles and must be engineered against opposite metrics. The training path is throughput-bound and latency-tolerant: it streams interaction events into the sharded embedding tables and the ranking model, it is measured in events processed per second and in how quickly a signal propagates, and a gradient applied a few seconds late costs almost nothing. The serving path is latency-bound and throughput-replicated: each request must finish inside the tens-of-milliseconds budget, it is measured in tail latency and in how many concurrent requests the fleet sustains, and being even a hundred milliseconds slow costs an impression. Conflating these two paths is a common design error, because the techniques that make training cheap, large asynchronous batches and relaxed deadlines, are exactly the techniques that ruin the serving path.

The two paths meet at two shared artifacts: the embedding tables and the feature store. The training path writes them; the serving path reads them. Freshness is precisely the rate at which updates flow from the streaming layer into the tables and features that the serving path is already querying, which is why neither can be a static snapshot but must be live, incrementally updated structures, a point Section 38.5 and Section 38.6 take up in detail. The concurrency the serving path must hold, the $C = \lambda L$ from Output 38.1.1, is the budget that Section 38.4 spends on replicated ranking. Keeping the two paths logically separate while letting them share exactly those two artifacts is the architectural spine of the chapter, and Section 38.8 draws the full picture.

# pip install implicit scipy
import implicit
from scipy.sparse import csr_matrix

# user x item interaction matrix, small enough to live in RAM
interactions = csr_matrix(load_clicks())          # one machine, one process
model = implicit.als.AlternatingLeastSquares(factors=128)
model.fit(interactions)                            # learns ALL embeddings in memory
recs = model.recommend(user_id, interactions[user_id], N=20)  # scores every item

5. What This Chapter Builds, Section by Section Beginner

With the requirements fixed, the scale proven, and the axes mapped, the path through the chapter is set. Section 38.2 learns the sharded embeddings, the terabyte-scale model that the parameter servers of Chapter 11 hold across machines. Section 38.3 builds candidate generation, the approximate-nearest-neighbor retrieval that takes the $200{,}000\times$ first cut of the funnel from Output 38.1.1. Section 38.4 ranks the survivors with a replicated deep scorer, spending the concurrency budget $C$ on distributed inference. Section 38.5 builds the feature store that both paths read and write. Section 38.6 adds real-time personalization, turning the click stream into fresh signals within seconds. Section 38.7 closes the loop with online evaluation, A/B testing, interleaving, and counterfactual estimates that judge served slates. Section 38.8 assembles the end-to-end architecture, and Section 38.9 turns it into a buildable project. Each section opens with the slice of Figure 38.1.1 it owns and the requirement from Section 1 it must satisfy.