Design a web-scale search engine (crawler and query service)

Questions first: is this a general web search or a vertical one? Do results need freshness measured in minutes (news) or days? Are images, ads and personalization in scope? What is the language mix? What latency and availability do queries need, and how stale may the index be?

Assumed scope: general web text search. In scope: crawling, parsing, link graph, index build, serving, ranking in two stages, freshness. Out of scope: ads, personalization, image and video search, and the details of spam classifiers.

Planning figures for the exercise, not facts about any real search engine.

• Crawl rate: 10 billion pages ÷ 30 days = 333 million a day. 333,000,000 ÷ 86,400 is about 3,900 pages a second.
• Bandwidth: 3,900 × 100 KB = 390 MB/s, about 3.1 Gbps. At 30 pages a second per node (the crawler chapter's figure), 3,900 ÷ 30 = 130 crawler nodes.
• Raw storage: 10 billion × 100 KB = 1 PB of HTML, 250 TB at 4:1 compression.
• Index size: assume 5 KB per page after text extraction and compression: 10 billion × 5 KB = 50 TB. At 50 GB a shard that is 1,000 shards of 10 million pages each.
• Query load if every shard served every query: each shard replica handles an assumed 250 queries a second, so 5,000 ÷ 250 = 20 replicas per shard, and 1,000 × 20 = 20,000 shard replicas. About 5,000 nodes at four shards each.
• With tiering: the top 10% of pages by quality sit in 100 shards that see all 5,000 q/s (100 × 20 = 2,000 replicas). Assume 10% of queries fall through to the other 900 shards: 500 q/s ÷ 250 = 2 replicas each, 1,800 replicas. Total 3,800, about 950 nodes. That is 5× fewer.
• Latency budget (300 ms): 20 query parsing, 120 shard search, 20 merge, 60 re-rank, 50 snippets, 30 slack.

The conclusion: the index must be sharded, replicated and tiered, and the fall-through rate decides the cost.

Pipeline.

1. Crawl. As in the crawler chapter: a frontier with priority queues in front and per-host politeness queues behind, robots and DNS caches, URL normalization, a URL-seen set, content fingerprints for duplicates, and sharding by host so one node owns one host's politeness. Refresh is a budget: pages that change often or matter more are revisited first.
2. Store. Fetched pages go to a blob or wide-column store keyed by hash of the normalized URL, with fetch time and a content hash. A "page fetched" event goes to a log.
3. Parse. Workers extract text, title, language, canonical URL and outgoing links; links go to the link graph store (edges, source to target) and to the frontier.
4. Offline jobs. A batch job computes link-based quality scores over the graph, and a build job turns documents into immutable index segments.
5. Serve. Segments are published to the serving shards. The index is derived data: it can always be rebuilt from the page store.

Data model. The index per shard holds, for each term, a postings list of (doc id, term frequency, positions), plus a forward store of doc id to title, URL and snippet source, and a static score. Doc ids are local to the shard. Documents are assigned to a tier by quality score and to a shard within the tier by hash of URL.

Document-partitioned vs term-partitioned index. I partition by document: each shard indexes a slice of pages completely, so each does the full query locally and returns its top 100. A term-partitioned index would ship long postings lists across the network for multi-word queries. The cost of document partitioning is that every query fans out to every shard in its tier.

Query path.

1. The query service normalizes and spell-corrects the query, then checks a result cache keyed by the normalized query with a short TTL (popular queries repeat a lot; measure the hit rate).
2. It fans out to one replica of each tier-1 shard, scoring with BM25 plus the static quality score. Late replicas are cut off at a deadline, and a slow request is hedged to a second replica.
3. It merges the shard top lists and re-ranks the best few hundred with a richer model (freshness, anchor text, language match). If tier 1 returns too few good results, it falls through to tier 2.
4. It builds snippets from the forward store.

Hard problem 1: tail latency. A query is as slow as its slowest shard out of 100. Use per-shard deadlines, hedged requests, and return a partial result rather than wait. Hard problem 2: freshness. Rebuilding 50 TB constantly is wasteful, so keep a small, frequently rebuilt fresh index for recently changed pages that is searched alongside the main one, with deletions recorded in a tombstone list. Merge fresh data into the main segments on a slower schedule.

• Document vs term partitioning. Document partitioning keeps each query's work local but fans out widely; term partitioning contacts few shards but moves large postings lists. Pick document partitioning for multi-word queries; flip only if queries are mostly single rare terms.
• Tiered vs flat index. Tiering cuts serving cost about 5× in the numbers above, at the price of a quality classifier and worse recall when good pages sit in the lower tier. Flip to flat if the cluster is small enough that cost does not matter.
• Batch rebuild vs incremental update. Batch is simple and gives clean immutable segments; incremental gives freshness. Use a small incremental fresh tier on top of batch, and widen it only when freshness is a product requirement (news).
• More replicas vs hedging. Extra replicas cost hardware; hedged requests cost extra load (a few percent) but cut the tail. Hedge first, add replicas when load, not tail, is the limit.

• Designing only the crawler or only the query service, when the question is both and the index build is the bridge.
• Putting the pages in one database and searching with LIKE, or putting the whole index on one machine.
• Ignoring politeness, robots rules and spider traps, or using a global queue with no per-host control.
• Forgetting fan-out arithmetic: sharding the index without noticing that every query now touches every shard.
• No ranking story beyond term matching, or no answer for pages that change.
• Updating the index synchronously on every crawl instead of building immutable segments.
• Skipping duplicate detection, so the results fill with copies of the same page.