Design a Search Engine

A web search engine accepts a text query and returns a ranked list of relevant pages from a corpus of billions of documents. The system has two halves. The offline half crawls the web and builds an index. The online half answers queries from that index within a few hundred milliseconds. This article covers the end-to-end design; the crawler and the index structure are treated at the level needed to reason about the whole system.

Requirements

Functional

• Crawl web pages and keep them reasonably fresh.
• Accept a free-text query and return the top results (title, URL, snippet), paginated.
• Rank results by relevance and page quality.
• Support query suggestions and spelling correction (optional, out of scope for the core design).

Non-functional

• Low query latency: p99 in the low hundreds of milliseconds.
• High availability of the query path. Stale results are acceptable; no results are not.
• Scalable to tens of billions of documents.
• Freshness: news pages within minutes, most pages within days.
• Politeness: the crawler must respect robots.txt and per-host rate limits.

Estimation

Assumptions: 50 billion indexed pages, 100 KB average raw page, 5 billion queries per day.

• Raw page storage: 50 x 10^9 x 100 KB = 5 x 10^15 bytes = 5 PB.
• Query rate: 5 x 10^9 / 86,400 ≈ 58,000 QPS average. Peak at 3x is about 175,000 QPS.
• Crawl rate to refresh the corpus on average every 30 days: 50 x 10^9 / (30 x 86,400) ≈ 19,000 pages per second, or about 1.9 GB/s of download bandwidth.
• Index size: the inverted index stores compressed text terms and positions, typically a fraction of raw text size. Assume 10% of extracted text. If extracted text is 20 KB per page (1 PB total), the index is roughly 100 TB, before replication.
• Serving: with 100 TB of index and, say, 1 TB of index served from memory or SSD per machine, one full copy needs on the order of 100 machines. Query load is served by replicating this set. At 175,000 QPS and roughly 1,000 QPS per replica machine, about 175 replicas of each shard are needed, because with document partitioning every query is sent to every shard. That is roughly 100 x 175 = 17,500 index machines. Real systems keep the hot portion in RAM.

What each number implies:

• 58,000 QPS average: the head of the query distribution must be served from a result cache. Assume (not measured) that 30 to 50% of queries repeat within minutes; each hit removes a full fan-out.
• About 100 shards per index copy (1 TB each): with document partitioning each query becomes about 100 shard requests, so 175,000 QPS produces about 17.5 million shard requests per second. Per-request cost and tail latency matter more than coordinator cost.
• 19,000 pages/s crawled: thousands of crawler machines and a frontier that spreads requests over millions of hosts, since per-host politeness (for example one request per second per host) caps how fast any single site can be fetched.
• 5 PB raw, 100 TB index: raw pages go to a distributed blob store; only the compressed index needs low-latency storage.

API design

GET /search?q=<query>&page=<n>&lang=<code>&safe=<bool>

Response:

{
  "results": [
    {"url": "https://example.com/a", "title": "...", "snippet": "...", "score": 0.87}
  ],
  "total_estimate": 1240000,
  "next_page_token": "..."
}

Internal APIs: the crawler pushes fetched documents to a document store; the indexer reads from it; index shards expose Search(query_terms, k) -> [(doc_id, score)] to the query coordinator.

Data model

The inverted index is the core structure. It maps each term to a posting list: the sorted list of document IDs containing that term. Lists are compressed with delta encoding (store gaps between sorted IDs) plus variable-byte or bit-packed codes. Index segments are immutable, which allows lock-free reads and simple caching.

High-level design

flowchart LR
    Web[(Web)] --> Crawler[Crawler fleet]
    Frontier[URL frontier] --> Crawler
    Crawler --> Docs[(Document store)]
    Crawler -->|new links| Frontier
    Docs --> Parser[Parser and dedup]
    Parser --> Indexer[Index builder]
    Docs --> Links[Link graph]
    Links --> Rank[PageRank batch job]
    Rank --> Indexer
    Indexer --> Shards[(Index shards)]
    User[User] --> LB[Load balancer]
    LB --> Coord[Query coordinator]
    Coord --> Shards
    Coord --> Cache[(Result cache)]

Components

• Frontier and crawler. The frontier decides what to fetch next by priority and politeness. Crawler workers fetch, honor robots.txt, and write results to the document store. See design a web crawler.
• Parser and deduplicator. Extracts text, links, and metadata. Near-duplicate detection uses fingerprints such as SimHash to skip mirror pages.
• Index builder. Tokenizes text, builds posting lists, merges them into immutable segments. Runs as a batch or incremental pipeline.
• Link graph and ranking job. Computes link-based scores such as PageRank offline.
• Query coordinator. Parses the query, fans out to shards, merges results, fetches snippets.
• Index shards. Each holds the postings for a subset of documents.

Query flow:

sequenceDiagram
    participant U as User
    participant C as Coordinator
    participant S as Shard (many)
    participant D as Doc metadata
    U->>C: query "rust memory model"
    C->>C: normalize, spell-correct, cache lookup
    C->>S: top-k for terms (parallel)
    S-->>C: local top-k (doc_id, score)
    C->>C: merge to global top-k
    C->>D: fetch title, URL, snippet
    D-->>C: metadata
    C-->>U: ranked results

Deep dive

Index partitioning: by document or by term

Web-scale engines use document partitioning. Each shard computes its local top-k independently, so intersection happens against local data and the coordinator only merges small result sets. The cost is that every query fans out to all shards, so tail latency depends on the slowest shard. Mitigations: send a hedged request (a duplicate of the same request) to a second replica of the shard when the first has not answered after a short delay, and use whichever answer arrives first; return partial results if a shard misses its deadline.

Documents are also tiered. A small top tier of high-quality pages is searched first; lower tiers are consulted only if the top tier yields too few results.

Query processing and ranking

For a multi-term query, the shard intersects posting lists (AND semantics). Intersection of sorted lists is done by merging or by skipping ahead with skip pointers, which avoids scanning the full length of the longer list. Scoring combines several signals:

• Text relevance, commonly TF-IDF or BM25: term frequency in the document, weighted by how rare the term is across the corpus, with length normalization.
• Link-based authority, such as PageRank: a page's score is derived from the scores of pages linking to it, computed iteratively over the link graph. Each iteration gives every page a base score plus a share of the score of each page that links to it, divided by that linking page's number of outgoing links; iteration repeats until scores stop changing meaningfully.
• Other signals: term proximity from stored positions, whether the term appears in the title or anchor text, page freshness, language match, and machine-learned ranking features.

Ranking is done in stages. A cheap first-stage score (BM25 plus static quality) selects a few hundred to a few thousand candidates per shard. A more expensive learned model then re-ranks only the merged top candidates. This limits the cost of expensive features to a small set.

Freshness and incremental indexing

Rebuilding the whole index for each update is too slow. The usual structure is a large, immutable main index plus a small, frequently rebuilt delta index for recently crawled or changed pages. Queries search both and merge. Periodically the delta is merged into a new main segment (compaction, similar to an LSM tree; see storage engines). Deleted or changed pages are tracked with a deletion bitmap applied at query time. The frontier assigns recrawl frequency by observed change rate and page importance, so news sites are fetched far more often than static pages.

Latency budget and fan-out arithmetic

A 200 ms p99 budget is roughly: 10 ms parse, spell-correct and cache lookup; 60 to 100 ms for the shard fan-out; 20 ms for merge and re-ranking; 30 ms for snippets; the remainder as slack. Each shard returns only its local top-k. For page 1 (10 results), the global top 10 is contained in the union of every shard's top 10, so shards return 10 entries each and the coordinator merges about 1,000 candidates. Page n needs about 10n per shard, so deep pagination is expensive; cap it (for example at page 10 to 20) and use next_page_token.

Hedging: send the duplicate request to another replica after the shard's p95 latency (for example 30 ms). That adds about 5% extra load and cuts the tail sharply. Set a hard shard deadline (for example 100 ms) and merge whatever has arrived; losing one shard of 100 drops about 1% of documents from that result set, an acceptable degradation.

Alternative to flat fan-out: an aggregation tree (coordinator, then mid-level aggregators each covering 10 to 20 shards) reduces connections per coordinator and merge work, at the cost of one more hop.

Caching

Query distributions are heavily skewed: a small fraction of distinct queries accounts for a large share of traffic. A result cache keyed by normalized query serves these without touching shards, with a short TTL to preserve freshness. A second cache holds hot posting lists in memory on shards. See caching fundamentals.

Bottlenecks and scaling

• Fan-out tail latency. With N shards, the query latency is the maximum shard latency. Use hedged requests, deadlines, and partial results.
• Hot queries. Result caching and replicated shards absorb spikes (breaking news).
• Index build time. Parallelize per shard with batch frameworks; build new index generations offline and swap atomically.
• Storage growth. Compress postings, tier documents, and drop low-quality pages from the index.
• Availability. Each shard has multiple replicas across zones; the coordinator routes around failed replicas. Replicate the index across regions and route users to the nearest.

Failure modes

• Replica slow or down. Hedging and health-based routing hide it; capacity headroom (for example 30%) must exist so that losing a replica or a zone does not overload the rest.
• All replicas of one shard lost. Queries succeed with partial results (that shard's documents are missing). Rebuild the shard from the index-generation files in blob storage.
• Bad index build or bad ranking model. Index generations are versioned and swapped atomically per shard group; keep the previous generation for instant rollback, and canary a new generation on a small share of traffic first.
• Crawler traps and hostile hosts. Infinite URL spaces (calendars, session IDs) are bounded by per-host page caps, URL depth limits, and duplicate-content fingerprints; slow hosts are backed off exponentially.
• Query of death or hot query. A pathological query is cut by a per-query CPU budget and early termination; breaking-news spikes hit the result cache with a short TTL (tens of seconds for news, minutes otherwise).
• At 10x scale (500 billion pages): about 1,000 shards per copy, so fan-out to all shards makes the tail dominate. Route queries only to relevant tiers, use an aggregation tree, and keep low-quality tiers out of the default path. Index build and crawl bandwidth (about 19 GB/s) also become the limiting resources.

Trade-offs and follow-up questions

• Freshness vs cost. More frequent recrawls improve freshness but consume bandwidth and crawler capacity. Prioritize by change rate and importance.
• Recall vs latency. Searching more tiers and deeper into posting lists finds more matches but raises latency. Early termination on sorted-by-quality lists trades some recall for speed.
• Follow-ups: How would autocomplete work? (See search autocomplete.) How would spam and link farms be handled? How would personalization be added without fragmenting the result cache? How would phrase queries be supported? (Store term positions in postings and verify adjacency.)
• What if a query contains very common words, such as "the who"? Their posting lists cover most documents. Use skip pointers, process the rarest term first, and keep impact-ordered (highest score first) lists so evaluation can stop early once the top-k cannot change.
• What if one shard is 10x hotter than the others? With random or hash-based document assignment, shards stay balanced; a hot shard usually means skewed assignment (for example by domain). Assign by document hash, not by site.
• What if the corpus doubles? Add shards for new documents and leave old shards unchanged, since document partitioning needs no resharding of existing data; only fan-out grows.
• How do you avoid serving a deleted page? Deletion bitmaps apply at query time, and the result cache TTL bounds how long a stale entry can survive.