Design a Web Search Engine
Problem
Design a general web search engine. It must crawl the public web, about 50 billion pages, build an index over them, and answer keyword queries with a ranked list of 10 results in under 300 ms. Popular pages should be re-crawled often enough that news appears within minutes while the long tail is refreshed every few weeks. The engine serves 100,000 queries per second at peak.
- Do we need images, video, or only HTML documents?
- How many languages should ranking handle well at launch?
- Is spell correction and query expansion in scope?
- Do we have to respect robots.txt and per-site crawl rate limits? (Assume yes.)
- What freshness is expected for news sites compared with ordinary pages?
- Do we need to render JavaScript-heavy pages to see their content?
Items in bold are existing requirements. The rest are suggestions. Confirm with the interviewer whether they are relevant first.
- Crawl the web continuously, discovering new URLs from links.
- Index page text so any keyword query can be matched.
- Return the 10 most relevant results with title, URL, and snippet.
- Recrawl pages at a rate proportional to how often they change and how important they are.
- Detect and collapse duplicate and near-duplicate pages.
- Demote spam and link farms.
- Correct obvious spelling mistakes in queries.
Items in bold are existing requirements. The rest are suggestions. Confirm with the interviewer whether they are relevant first.
- 50 billion indexed pages; at about 100 KB of HTML and 10 KB of extracted text each, that is 5 PB raw and 500 TB of text.
- Query latency under 300 ms at p99 for 100,000 queries per second at peak.
- The inverted index is roughly 30% of text size after compression, about 150 TB, held mostly in memory and SSD across thousands of machines.
- Recrawling 50 billion pages every 30 days requires about 20,000 fetches per second; important pages are fetched far more often.
- Crawl politely: never exceed a site's crawl delay or open many connections to one host.
- Serving stays available when a whole index shard replica fails.
A distributed crawler fetches pages under per-host politeness rules and stores raw content. An indexing pipeline extracts text and links, deduplicates content, computes link-based quality scores, and builds an inverted index partitioned by document across many shards. At query time a root server fans the query out to every shard, each returns its best candidates using a cheap scoring function, and a second stage reranks the merged top few hundred with a more expensive model.
Loading diagram...
Crawl. The frontier hands each fetcher a batch of URLs grouped by host, already scheduled to respect politeness. The fetcher resolves DNS from a local cache, checks the cached robots.txt, downloads the page with conditional headers (If-Modified-Since), and writes the response to the raw page store.
Process. The parser extracts text, title, language, and outgoing links. It computes a content fingerprint (SimHash) to detect near-duplicates and keeps one canonical document per cluster. New links are normalized and sent to the frontier with a priority based on the linking page's quality.
Index. Periodically, and continuously for a fresh tier, the index builder turns documents into posting lists (term -> [doc id, positions, field flags]), attaches static quality scores, and publishes new index segments to the shard servers.
Query. The frontend normalizes and spell checks the query, then the root server sends it to one replica of every shard. Each shard intersects the posting lists, scores matches with a fast function (BM25 plus quality), and returns its top 50. The root merges results, sends the top few hundred to the reranker, fetches snippets for the final 10, and returns the page.
Decides what to fetch next and fetches it without overloading any website.
- Fetchers in Go with asynchronous HTTP, a shared DNS cache, and a robots.txt cache.
- The frontier as a sharded priority queue persisted in a key-value store such as RocksDB.
Diagram
Loading diagram...
Data Model
url_state(url_hash, url, host, last_fetched, last_changed, change_rate, priority, next_fetch_at)
Scalability
Partition the frontier by host hash so all URLs of a host live on one frontier shard, which can then enforce the host's crawl delay locally without coordination. A Bloom filter in front of the seen-URL table rejects most already-known URLs cheaply.
Tradeoffs
Separating front queues (priority) from back queues (one per host) lets important pages jump ahead while still guaranteeing politeness. A single global priority queue would keep hammering a host that happens to have many important URLs.
Extracts clean content, detects duplicates, and builds the link graph.
- A batch and streaming framework such as Spark or Beam.
- SimHash fingerprints for near-duplicate detection; a headless browser pool for the small share of pages that need JavaScript rendering.
Data Model
documents(doc_id, canonical_url, title, text, language, fingerprint, quality, fetched_at)links(from_doc_id, to_doc_id, anchor_text)
Scalability
Fingerprints are bucketed by bit bands so only documents sharing a band are compared, avoiding all-pairs comparisons. Rendering JavaScript is expensive, so it runs only for hosts known to need it.
Tradeoffs
Anchor text from incoming links is indexed as part of the target document because it often describes a page better than the page itself. This makes the index depend on the link graph, so index builds wait on the link pass, which is acceptable for the main tier.
Hold posting lists for a slice of the corpus and return top candidates for a query.
- Custom C++ servers with compressed posting lists (variable-byte or PForDelta encoding), memory for hot terms and SSD for the rest.
Data Model
- Posting list:
term -> [(doc_id, term_frequency, positions, field_flags)]sorted by doc id, with skip pointers. - Per document: static quality score and length used during scoring.
Scalability
Partition by document: 50 billion pages over about 2,000 shards is 25 million documents per shard. Each shard is replicated several times; adding replicas adds query throughput. Documents are ordered by quality inside each shard so a shard can stop scanning early once it has enough strong candidates.
Tradeoffs
Document partitioning means every query touches every shard, but each shard does independent work and a slow shard can be skipped with little quality loss. Term partitioning would touch fewer machines per query but requires shipping huge posting lists between machines for multi-term queries and creates hot shards for common words.
Turns candidates into the final ordered list.
- First stage: BM25 combined with static quality, computed inside the shard.
- Second stage: a learning to rank model (gradient boosted trees or a small transformer) running on a dedicated fleet.
Scalability
The expensive model sees only a few hundred documents per query, which bounds its cost regardless of corpus size. Results for popular queries are cached at the root for a few minutes.
Tradeoffs
A two-stage cascade spends compute where it matters. Running the expensive model on every matching document would be impossible within the latency budget, while using only BM25 would rank poorly for ambiguous queries.
Computes a query-independent importance score for every page from the link graph.
- An iterative graph computation on Spark or a dedicated graph engine, run daily.
Scalability
The graph has hundreds of billions of edges, so the job partitions vertices across workers and runs a fixed number of power iterations, warm-started from the previous day's scores so it converges quickly.
Tradeoffs
A PageRank-style score is cheap to apply at query time and resists manipulation better than raw inbound link counts. It is easy to game with link farms, so spam detection discounts links from clusters of low-quality sites.