Project Extension

"I was a tidy matrix factorization over a thousand movies. Then someone handed me the billion-item catalog, and I had to learn that I would never again fit on one machine."

A Toy Recommender, Dreaming of the Billion-Item Catalog

A recommender read passively teaches the shape of a retrieve-then-rank funnel; a recommender rebuilt teaches why each stage of that funnel had to be distributed. The eight sections before this one walked the system as a finished artifact: a sharded embedding table feeding a sharded candidate retriever, a distributed ranker scoring the candidates, a feature store serving point-in-time-correct features, a bandit layer adapting in real time, and an A/B harness proving the whole thing lifts a business metric. This section inverts that posture. It gives you a staged construction plan that begins with a baseline small enough to fit one laptop and honest enough to measure against, then removes one single-machine assumption per stage. Each stage draws on a specific earlier section and a specific earlier chapter, so the project doubles as a guided tour back through the book: when you shard the embedding table you are applying the parameter-server and sharded-embedding machinery of Chapter 11, when you build the retriever you are applying the distributed vector search of Chapter 25, when you train the ranker you are applying the data parallelism of Chapter 15, and when you add the bandit you are applying the online-learning and streaming machinery of Chapter 9. The discipline that makes the project worth its time is the one Section 1.1 opened with: distribute a stage only when a ceiling forces it, and prove with a number that the distribution helped.

1. The Baseline You Scale Out From Beginner

Every honest scale-out project begins with a single-machine baseline, for two reasons. The first is correctness: the distributed retriever must return essentially the same items as the brute-force baseline, and you cannot check that against a baseline you never built. The second is measurement: speedup, recall, and lift are all defined relative to a reference, and the reference is the one-machine system. Your baseline is a single process over a public interaction dataset at MovieLens scale (a few thousand users, a few thousand items, a few hundred thousand ratings) that learns a low-rank factorization $R \approx P Q^{\top}$, where $P$ holds a $K$-dimensional embedding per user and $Q$ a $K$-dimensional embedding per item, then retrieves the top items for a user by scoring every item and keeping the largest dot products. This is the two-tower model in its smallest honest form, and it fits one machine comfortably, which is exactly the point: it is the thing the rest of the project will outgrow.

The code below is that baseline compressed to one dependency-free file. It generates synthetic implicit interactions from latent user and item tastes, trains the factorization with a Bayesian-personalized-ranking objective, retrieves the exact top items per user, and then immediately shows the second scale-out move, replacing the brute-force scan with a clustered approximate-nearest-neighbor search (Section 38.3) that probes only a few item clusters, and measures how much recall it keeps against the exact retrieval. That recall-against-exact comparison, the retrieval analogue of the gradient identity in Section 1.1, is the invariant your project tracks at every retrieval stage.

import math, random

random.seed(0)
N_USERS, N_ITEMS, DIM = 200, 500, 16

def rand_vec(seed, dim):              # deterministic latent vector from a string seed
    r = random.Random(seed)
    return [r.gauss(0, 1) for _ in range(dim)]

user_lat = {u: rand_vec(f"u{u}", DIM) for u in range(N_USERS)}
item_lat = {i: rand_vec(f"i{i}", DIM) for i in range(N_ITEMS)}

def dot(a, b):
    return sum(x * y for x, y in zip(a, b))

# Synthetic implicit feedback: each user "engaged" with its true top-10 items.
interactions = {}
for u in range(N_USERS):
    s = [dot(user_lat[u], item_lat[i]) for i in range(N_ITEMS)]
    order = sorted(range(N_ITEMS), key=lambda i: s[i], reverse=True)
    interactions[u] = set(order[:10])

# Stage A: single-machine matrix factorization trained by BPR (two-tower).
K, lr, reg, EPOCHS = 8, 0.05, 0.01, 40
P = {u: [random.gauss(0, 0.1) for _ in range(K)] for u in range(N_USERS)}
Q = {i: [random.gauss(0, 0.1) for _ in range(K)] for i in range(N_ITEMS)}
pairs = [(u, i) for u in range(N_USERS) for i in interactions[u]]
for _ in range(EPOCHS):
    random.shuffle(pairs)
    for u, i in pairs:
        j = random.randrange(N_ITEMS)                 # sample a negative item
        while j in interactions[u]:
            j = random.randrange(N_ITEMS)
        pu, qi, qj = P[u], Q[i], Q[j]
        g = 1.0 / (1.0 + math.exp(dot(pu, qi) - dot(pu, qj)))   # BPR gradient weight
        for k in range(K):
            pu[k] += lr * (g * (qi[k] - qj[k]) - reg * pu[k])
            qi[k] += lr * (g * pu[k] - reg * qi[k])
            qj[k] += lr * (-g * pu[k] - reg * qj[k])

# Stage B: exact top-k retrieval (the brute-force baseline).
def exact_topk(u, k):
    return sorted(range(N_ITEMS), key=lambda i: dot(P[u], Q[i]), reverse=True)[:k]

# Stage C: sharded ANN-style retrieval. Cluster items, probe the nearest clusters.
NCLUST, NPROBE = 16, 6
centroids = [rand_vec(f"c{c}", K) for c in range(NCLUST)]
for _ in range(3):                                    # a few k-means refinement passes
    asg = {c: [] for c in range(NCLUST)}
    for i in range(N_ITEMS):
        asg[max(range(NCLUST), key=lambda c: dot(centroids[c], Q[i]))].append(i)
    for c in range(NCLUST):
        if asg[c]:
            centroids[c] = [sum(Q[i][k] for i in asg[c]) / len(asg[c]) for k in range(K)]
clusters = {c: [] for c in range(NCLUST)}
for i in range(N_ITEMS):
    clusters[max(range(NCLUST), key=lambda c: dot(centroids[c], Q[i]))].append(i)

def ann_topk(u, k):                                   # scan only the probed shards
    probe = sorted(range(NCLUST), key=lambda c: dot(centroids[c], P[u]), reverse=True)[:NPROBE]
    cand = [i for c in probe for i in clusters[c]]
    return sorted(cand, key=lambda i: dot(P[u], Q[i]), reverse=True)[:k]

# Evaluation: MF recall vs ground truth, and ANN recall vs the exact retrieval.
def recall(pred, truth):
    return len(set(pred) & set(truth)) / len(truth)

KK, mf_recall, ann_vs_exact, scanned = 10, 0.0, 0.0, 0
for u in range(N_USERS):
    ex, an = exact_topk(u, KK), ann_topk(u, KK)
    mf_recall += recall(ex, interactions[u])
    ann_vs_exact += recall(an, ex)
    probe = sorted(range(NCLUST), key=lambda c: dot(centroids[c], P[u]), reverse=True)[:NPROBE]
    scanned += sum(len(clusters[c]) for c in probe)
mf_recall /= N_USERS; ann_vs_exact /= N_USERS
frac = scanned / (N_USERS * N_ITEMS)

print(f"users / items / latent K        : {N_USERS} / {N_ITEMS} / {K}")
print(f"MF recall@{KK} vs ground truth    : {mf_recall:.3f}")
print(f"ANN recall@{KK} vs exact retrieval: {ann_vs_exact:.3f}")
print(f"fraction of catalog scanned      : {frac:.3f}  (nprobe={NPROBE}/{NCLUST})")
print(f"retrieval speedup vs exact scan  : {1.0 / frac:.1f}x")

users / items / latent K        : 200 / 500 / 8
MF recall@10 vs ground truth    : 0.441
ANN recall@10 vs exact retrieval: 0.961
fraction of catalog scanned      : 0.416  (nprobe=6/16)
retrieval speedup vs exact scan  : 2.4x

2. Staging the Scale-Out, Milestone by Milestone Intermediate

With the baseline in hand, you scale the recommender out one stage at a time, in the order of Figure 38.9.1, never advancing until the current stage hits its milestone. The discipline of one-stage-at-a-time matters because it isolates cause and effect: when recall drops or tail latency spikes, exactly one thing changed, and you know which section's mechanism to inspect. Table 38.9.1 is the project plan. Each row names the stage, the binding ceiling that forces the distribution, the section and chapter that supply the mechanism, and the quantitative milestone that tells you the stage is done.

The milestones are quantitative on purpose. "Sharded the embedding table" is not a milestone; "the table now spans four nodes and per-key lookups return in under a millisecond at the p99" is. Stage 1 is a capacity stage, judged by the catalog size it unlocks once the table no longer fits one box, the embeddings-at-scale move that Chapter 11 owns. Stage 2 is a quality-under-latency stage, where the move from exact scan to approximate nearest neighbor (Chapter 12, Chapter 25) trades a sliver of recall for a large latency win, and the milestone forbids trading away more than a point. Stage 3 is a model-quality stage, where data-parallel training (Chapter 15) must reproduce the single-node ranking AUC. Stage 4 is a correctness stage: a feature store with point-in-time joins exists to forbid the most common silent bug in recommendation, training on a feature value that was not yet known at serving time. Stage 5 is an online-lift stage, where a bandit (Chapter 9) earns its place only if it raises reward over the static ranker by a statistically significant margin. Stage 6 closes the loop with the A/B harness and the service architecture, proving the fully distributed system meets its tail-latency SLO at a defensible cost per thousand requests, the evaluation discipline of Chapter 5 applied to a live recommender.

3. The Numbers Your Project Must Hit Intermediate

A recommendation project lives or dies by whether its claimed wins are real, so each milestone is a quantity you compute, not a feeling. Retrieval quality is recall at $k$, the fraction of the relevant items that appear in the returned top $k$,

where the ranking AUC is the probability that the ranker scores a clicked item above a non-clicked one, the standard pairwise quality metric for the Stage-3 ranker. For Stage 2, the milestone pins approximate recall against exact recall to within a few points, exactly the gap Output 38.9.1 measured at $0.961$ before any latency pressure; the retrieval speedup is the reciprocal of the catalog fraction you scan, so probing $42\%$ of clusters buys the $2.4\times$ of that output. For Stage 1, the relevant number is sizing the sharded embedding table, the parameter ceiling that defines this case study. A catalog of $N$ items with a $K$-dimensional float32 embedding needs $4 N K$ bytes of item-tower parameters, and the shard count is $\lceil 4 N K / M \rceil$ for a per-node memory budget $M$. Concretely, $N = 10^9$ items at $K = 128$ is $4 \times 10^9 \times 128 = 512$ gigabytes of item embeddings alone, so with $M = 64$ gigabytes per node you need at least $\lceil 512 / 64 \rceil = 8$ shards before any replication, and the user tower adds its own table on top, which is the stage-1 capacity argument from Section 38.2 made arithmetic. The service-level milestones are tail latency and unit cost: p99 latency must sit under the serving SLO at the target queries per second, and cost per thousand requests must stay within budget, the two numbers that decide whether the system ships. Compute all of these targets before you build, so each milestone is a prediction you test rather than a result you rationalize.

One stage has a milestone that is not a number but a guarantee. The feature store of Stage 4 exists to make point-in-time joins correct: when you assemble the training row for an interaction at time $t$, every feature value in that row must be the value that was known at time $t$, never a value computed afterward. Violating this is training on the future, the leakage bug that makes an offline AUC look spectacular and an online deployment collapse. The milestone is therefore binary, zero leakage, and you verify it by replaying the feature timeline and asserting that no join reads a value with a timestamp later than the label it is paired with, the streaming-correctness discipline of Chapter 9 applied to the most expensive silent failure in recommendation.

# Each staged milestone -> the production tool that owns it.
STACK = {
    "shard embeddings":   "TorchRec           # sharded embedding tables across GPUs",
    "sharded retrieval":  "FAISS-IVF / ScaNN  # ANN shards, scatter-gather merge",
    "distributed ranker": "torch DDP / TorchRec DLRM  # data-parallel ranker train",
    "feature store (PIT)":"Feast / Tecton     # point-in-time-correct feature joins",
    "real-time bandit":   "Vowpal Wabbit / River  # online contextual bandit updates",
    "A/B + serving":      "Ray Serve + a stats harness  # replicate, route, evaluate",
}
for stage, tool in STACK.items():
    print(f"{stage:20s}-> {tool}")

4. Extension Challenges Worth the Cluster Advanced

Once the six stages hit their milestones you have a working distributed recommender, and the project becomes a platform for the harder questions the chapter only gestured at. Each extension below adds one capability that a real billion-item system needs, and each reaches into a different part of the book, so finishing them turns the capstone from a funnel into a system. Add multi-objective ranking, where the ranker no longer optimizes click probability alone but a weighted blend of engagement, dwell time, and a long-term-value proxy, and measure the trade-off frontier as you sweep the objective weights; the milestone is a Pareto curve, not a single AUC. Tackle cold-start, the problem of users and items with no interaction history and therefore no learned embedding, by adding content features (a side tower over item metadata) that let the model place a brand-new item near similar ones; the milestone is recall on a held-out set of items the model never saw during training. Add a diversity-and-fairness pass over the final ranking, so that a single popular category cannot monopolize the slate and so that exposure is allocated across item providers within a stated bound; the milestone is a diversity or exposure metric held at a target while recall stays within budget.

Two further extensions push the candidate generator and the trust boundary. Add graph candidates, generating a slice of the retrieval set from a user-item interaction graph rather than embedding similarity alone, the distributed-graph-ML machinery of Chapter 13; co-visitation and random-walk candidates often surface long-tail items that pure embedding retrieval misses, and the milestone is a long-tail-coverage lift. If your recommender serves multiple tenants or markets and you aggregate signals across them, add the reliability and privacy guarantees of Part VII so that one corrupted upstream signal cannot poison the slate and no single user's interactions leak through a cross-tenant aggregate, the robust-aggregation and differential-privacy ideas of Chapter 35. Each extension is a small, bounded change to a working system, which is exactly the posture in which distributed-systems concepts are learned best: against a baseline you can measure, in a funnel you already understand.

5. Chapter Summary and What You Built Beginner

This section closes Chapter 38, so it is worth stating the through-line the whole chapter built. We began with the problem definition (Section 38.1): recommend from a catalog so large that the model itself, the embedding table mapping every user and item to a vector, does not fit on one machine. From there the chapter walked the retrieve-then-rank funnel stage by stage, and every stage was the same move applied to a different part of the system, partition the work across machines, move only what must be moved between them, and recombine the result correctly. Sharding the embedding table (Section 38.2) distributed the model itself, applying the sharded-embedding and parameter-server machinery of Chapter 11 to the parameters that no node could hold. Sharded candidate retrieval (Section 38.3) turned a query into the scatter-gather approximate-nearest-neighbor pattern of Chapter 25. The distributed ranker (Section 38.4) applied the data parallelism of Chapter 15 to score candidates at scale. The feature store with point-in-time joins (Section 38.5) and the real-time bandit (Section 38.6) brought the streaming-correctness and online-learning machinery of Chapter 9 to bear, and the A/B harness (Section 38.7) and service architecture (Section 38.8) proved, with the evaluation discipline of Chapter 5, that the distributed system lifts a real metric without breaking its latency budget. The chapter is, end to end, one distributed system whose distinguishing feature is that the model, not just the data, is too big for one machine.