Topic 11.7
Scaling
In one line
When the corpus or the traffic outgrows one machine, you split the index into shards (to hold more) and copy it into replicas (to answer more, and to survive failures). Every query then goes to every shard, and the results are merged.
Think of it like this
A library with a million books cannot fit in one room, so it opens three branches, each holding a third. To find the newest cookery books a clerk phones all three branches, gets each one's best picks and chooses the overall best. That is sharding. If the queue at each branch is too long, each branch hires a second clerk who works from a photocopy of the same catalogue. That is replication.
Words you'll meet
New words in this topic, in plain English. Come back here whenever one feels fuzzy.
- Shard
- One part of the index. Each shard holds a share of the documents.
- Replica
- An extra copy of a shard that answers questions in parallel and takes over if another copy fails.
- Scatter-gather
- Send the query to every shard, then gather and merge their answers.
- Coordinator
- The part that fans a query out to shards and merges the results.
- Consistent hashing
- A way to assign data to machines so that adding or removing one moves only a small share.
- Backlog
- Work waiting in a queue, such as documents waiting to be indexed.
- Tenant
- One customer whose data must stay separate from other customers' data.
- Headroom
- Spare capacity kept so a failure or a spike does not overload the rest.
Step by step
01Step 1: decide if you need to shard at all
One machine with enough memory is simpler and faster than any cluster. Shard when the index does not fit, or when one search is too slow on a single machine. If the problem is only traffic, replicas are enough.
02Step 2: assign documents to shards
Use a stable function of the document id. In the example the 8 documents fall as shard 0 = D1, D4, D10; shard 1 = D2, D8; shard 2 = D3, D6, D12. Chunks of one document should stay together, so hash the document id and not the chunk id.
def shard_of(doc_id):
return sum(ord(c) for c in doc_id) % SHARDS03Step 3: query every shard for its top n
Each shard answers from its own slice. Make sure n is at least k. Asking for fewer is the quiet bug in the example's second run.
04Step 4: merge and compare with the single index
Merge by sorting on (score descending, id). Compare with what one big index would return. The example's first run matches, the second does not. Keep this comparison as a test whenever you change the sharding.
05Step 5: add replicas for traffic and failure
Divide peak traffic by what one replica handles and round up, then add one for headroom. Spread replicas over different machines or zones, so one failure takes out at most one copy.
06Step 6: scale ingestion on the backlog
Watch queue length and the age of the oldest message. When the age passes your allowance (Topic 11.2), add workers, up to the limits of the embedding provider, which is usually the real bottleneck.
Try it yourself
- 1
Ask for more
Change the second run from 1 to 2 results per shard. Does the merged top 3 match the single index now? Why is
n = kalways enough? - 2
Change the shard count
Set
SHARDSto 2. Look at the first line: which documents moved? What would that mean for a live index? - 3
Size the cluster
Set the peak to 300 questions a second. How many replicas per shard, and how many machines in total?
Code & diagrams
Scores are shared-word counts (the course's toy retriever), which are comparable across shards. The shard rule is a toy hash.
Expected output
shards: 0=D1,D4,D10 1=D2,D8 2=D3,D6,D12
one big index, top 3: D10(7), D12(4), D6(3)
each shard returns its top 3:
shard 0 returns D10(7), D1(1), D4(1)
shard 1 returns D8(2), D2(0)
shard 2 returns D12(4), D6(3), D3(2)
merged top 3: D10(7), D12(4), D6(3) | same as one big index: yes
each shard returns its top 1:
shard 0 returns D10(7)
shard 1 returns D8(2)
shard 2 returns D12(4)
merged top 3: D10(7), D12(4), D8(2) | same as one big index: no
capacity: 120 questions a second, 45 per replica -> 4 replicas per shard (one spare), 12 machinesBreak it on purpose
Errors are the best teachers. Make each change, read the error, guess what went wrong, then reveal the answer.
Break #1
Change the number of shards
In the Python tab, change SHARDS = 3 (the line above QUESTION) to SHARDS = 4 and run.
Myth vs fact
Myth
More shards always make search faster.
Fact
Each query touches every shard and waits for the slowest one. More shards add merging work and more chances for a slow node. Shard for size, not for speed.
Myth
Replicas and shards are the same thing.
Fact
Shards split the data. Replicas copy it. Shards let you hold more; replicas let you answer more and survive failure.
Myth
Merging per-shard top-k lists is always safe.
Fact
It is exact only if each shard returns at least k results and the scores are comparable across shards. Keyword scores with local statistics are not.
Pro corner
Extra depth for experienced readers. New to this? Skip it for now and come back later.
- ▸
Tail latency gets worse with fan-out: a query is as slow as its slowest shard. With 20 shards, even a 1 percent chance that a shard is slow becomes an 18 percent chance that the query is slow. Use per-shard timeouts and, when a shard is late, return partial results with a flag. Hedged requests (sending to a second replica if the first is slow) trim the tail.
- ▸
Approximate indexes (Phase 3) lose recall (the share of the true best results they find) when split: each shard's graph or clusters see only part of the data. Measure recall after sharding, not before.
- ▸
Size a replica at your latency target, not at its maximum. The 45 questions a second in the example only count if that is the rate at which a replica still meets its p95 budget (Topic 11.4). Load-test while watching p95: a replica pushed to its breaking point answers more per second and every answer is late. Then divide peak traffic by the safe rate, not by the best rate.
- ▸
Rebalancing moves data while serving traffic. Do it by building the new layout beside the old one and switching, the same pattern as changing the model (Topic 11.8).
- ▸
Many vector databases (Phase 5) shard and replicate for you, and expose settings for shard count and replication factor. The principles above tell you what the settings mean and what they cost.
- ▸
System Design (
/system-design) covers sharding, replication and consistent hashing in general; this topic applies them to search.
Remember this
- 1
Why two different tools. Two different limits. Size: the index (vectors, text, metadata) may not fit in one machine's memory. Fix: shards, each holding a part of the data. Traffic: one copy can answer only so many questions per second. Fix: replicas, extra copies that each answer a share of the questions. They also give availability: if one machine dies, its twin keeps serving.
- 2
How documents get to shards. Use a stable rule:
shard = hash(doc_id) mod number_of_shards.modmeans the remainder after division. The example uses the sum of the characters' codes (a toy hash): D1 is the codes 68 and 49, which add to 117, and 117 mod 3 is 0, so D1 lives on shard 0. The same id always lands on the same shard, so an update or a delete goes to the right place. Real systems use proper hash functions, and often consistent hashing or many small "virtual" shards, so adding a machine moves only a small share of the data. The break-it shows what a naive mod does when the shard count changes. - 3
Scatter-gather. A question cannot know which shard holds the best passage, so it goes to every shard. Each returns its local top results, and a coordinator merges them into the global top k. The merge is simple: sort by score. It only gives the right answer if each shard returns at least k results. In the example each shard returning its top 3 matches the one big index (D10, D12, D6). With only its top 1, shard 2's second hit D6 (score 3) is lost and D8 (score 2) takes its place.
- 4
Scores must be comparable across shards. Cosine similarity is comparable everywhere. BM25 (Phase 1) is not: its idf values (how rare a word is, which decides how much it counts) depend on each shard's own statistics, so a rare word looks more important on a shard that has few such documents. Fix: compute statistics over the whole corpus and share them, or use a scoring method that does not depend on local statistics, or fuse by rank (Phase 6).
- 5
Capacity by hand. If peak traffic is 120 questions a second and one replica handles 45, you need
ceil(120 / 45) = 3replicas, plus one spare so a failure or a deploy does not overload the rest: 4. Because every question touches every shard, each shard needs its own full set of replicas: 3 shards × 4 replicas = 12 machines. Sharding is expensive in machines; do not shard before you must. - 6
Ingestion has its own scaling problem: the backlog. When thousands of documents arrive at once, the queue between "changed" and "indexed" grows. Watch the backlog size and the oldest item's age (that is your index lag, Topic 11.2). Scale workers on backlog, not on CPU, and rate-limit the embedding calls so ingestion cannot starve the query path. Many tenants (customers) are a third axis: either one index per tenant (strong isolation, many small indexes) or one shared index with a tenant filter (cheap, and a leak risk if the filter is missed, Topic 10.4).
Explain it without notes
What is the difference between sharding and replication, and which limit does each address?
How does scatter-gather search work, and when does it return the wrong top k?
How would you size a cluster for 120 questions a second when one replica handles 45?
Why does a RAG system's ingestion path need its own scaling plan?
Practice
Write shard_of(doc_id, shards) returning the sum of the character codes of doc_id modulo shards. Print the shard of D1, D2 and D3 for 3 shards, then for 4 shards.
Write merge(parts, k) taking a list of lists of (score, doc_id) pairs and returning the best k pairs by score (highest first, ties by id). Try it on two parts.
Trade-offs
- ↔
More shards: capacity for a larger index, higher tail latency and more machines.
- ↔
More replicas: more throughput and availability, paid for in machines and in keeping copies in sync.
- ↔
One index per tenant: strong isolation, more indexes to run; a shared index with a filter: cheaper, and a leak if the filter is missed.
Done when you can
Done when you can explain what shards and replicas are for.
Done when you can assign documents to shards with a stable rule and say why changing the count is costly.
Done when you can explain scatter-gather and why each shard must return at least k results.
Done when you can size replicas from traffic, with headroom.
Done when you can name the metric to scale ingestion on.