Design a Web Search Engine: Crawling, Inverted Indexes and Ranking

Typing a few words into a search box and getting relevant results from billions of pages in a fraction of a second is one of the great achievements of systems engineering. Interviewers sometimes ask "design a web search engine" or "design a web crawler" to see whether a candidate can decompose a huge problem into pipelines, choose partitioning strategies, and reason about latency across hundreds of machines.

This article designs an educational search engine over about a billion pages. It does not try to recreate any commercial search engine, whose ranking systems are vastly more complex and largely unpublished; it focuses on the architecture every search system shares. Google's founders' 1998 paper on their prototype remains a readable introduction to the components (The Anatomy of a Large-Scale Hypertextual Web Search Engine).

• How to estimate crawl, index and query workloads.
• A polite, distributed crawler: frontier, robots.txt, per-host limits, deduplication and traps.
• Inverted indexes, with a runnable example, and how to build them at scale.
• Document-partitioned shards, replicas and scatter-gather query serving.
• Controlling tail latency with timeouts, caching, early termination and hedged requests.
• Ranking in stages, freshness tiers, failures and safety.

Build a search engine that crawls a large part of the public web (about a billion pages), keeps the index reasonably fresh, and answers keyword queries with ten ranked results, each with a title, URL and snippet, quickly and reliably.

Functional requirements: crawl and refresh pages; build and update an index; answer queries with ranked results and snippets; remove pages on request.

Non-functional requirements

Corpus: 1 billion pagesRaw HTML: average 100 KB -> 100 TB (kept in object storage, compressed)Extracted text: average 10 KB -> 10 TBInverted index: assume about 30% of text size after compression -> about 3 TBDocument store for titles, URLs and snippets: about 10 TB (compressed text)Crawl: refresh everything every 30 days -> 33 million pages/day -> about 390 pages/s,  about 39 MB/s (0.3 Gbps) of downloads, more for the fresh tierQueries: 100 million/day -> about 1,160/s average, about 3,500/s peak

Conclusions: the index (terabytes) does not fit on one machine and must be partitioned; queries (thousands per second, each touching every partition) need replicas and careful latency control; crawling is a large but steady background pipeline whose hardest constraint is politeness, not bandwidth.

GET /v1/search?q=monsoon+trekking+kerala&page_token=...&lang=en200 { "results": [ { "url": "...", "title": "...", "snippet": "...", "published_at": "..." } , ... ],      "next_page_token": "...", "partial": false }Internal:  Frontier:   add_urls(urls, source, priority) ; lease_batch(host_bucket) -> urls to fetch  Indexer:    consume fetched documents -> build or update index segments  Shard RPC:  search_shard(query, k, deadline) -> top-k (doc_id, score)  Doc store:  get_snippets(doc_ids, query) -> titles, URLs, snippets

partial: true signals that some shards did not answer in time; the results are still useful.

URL frontier (partitioned by hash of host):  url_state (url_hash, url, host, priority, last_fetched_at, next_fetch_at, content_hash, status)  host_state (host, robots rules, crawl_delay, last_fetch_at, error_count)Fetched pages: object storage, key by url_hash + fetch time; metadata in a tableDocument table: (doc_id, url, title, language, length, static_quality_score, fetched_at)Inverted index (per shard, immutable segments plus a small mutable segment):  term -> postings list: [(doc_id, term_frequency, positions...), ...] sorted by doc_id  per-document data for scoring: length, field lengths, static qualityLink graph (for authority signals): (from_doc, to_doc), processed in batch

One crawler process --fetches--> pages --> relational database with full-text indexQuery --> SELECT ... WHERE text matches query ORDER BY rank LIMIT 10

A single crawler cannot fetch 33 million pages a day politely; a single database cannot hold a billion pages' index or answer thousands of queries per second; and database text search offers limited control over ranking at this scale.

1. Crawling politely at scale: millions of hosts with different limits and robots rules, plus traps and duplicates.
2. Index size: terabytes of postings.
3. Query fan-out: every query must consult the whole index, which lives on many machines.
4. Tail latency: with many shards, one slow shard slows every query.
5. Ranking quality: term matching alone ranks spam and thin pages highly.
6. Freshness: a monthly rebuild misses today's news.

Seed URLs + discovered links --> URL frontier (priority queues per host, partitioned by host)   --> Fetcher fleet (each fetcher owns a set of hosts)         - check robots.txt (cached per host) and crawl delay         - one or few connections per host at a time; respect rate limits         - fetch, follow limited redirects, store raw page in object storage   --> Parser: extract text, title, links, language; compute content hash   --> Dedup: exact (hash) and near-duplicate (shingles or similar fingerprints)   --> New links (normalised) back to the frontier; documents to the indexing pipeline

• Politeness: the Robots Exclusion Protocol, standardised as RFC 9309, lets site owners state which paths crawlers may fetch (RFC 9309). The crawler fetches and caches each host's robots.txt, obeys it, and limits its own request rate per host (for example, at most one request per second, slower if the site responds slowly or with errors).
• Partition by host: all URLs of a host go to one frontier partition and one fetcher, so per-host rate limits are enforced in one place without coordination.
• Prioritisation: important pages (many inbound links, popular sites) and frequently changing pages are fetched more often; new links get an initial priority from their source.
• URL normalisation (lower-case host, remove fragments, sort known tracking parameters) avoids fetching the same page under many URLs.
• Traps: calendars and generated URLs can create infinite link spaces; cap URLs per host, depth and URL length, and detect repeating patterns.
• Change detection: compare content hashes; unchanged pages need no re-indexing, and their refresh interval can lengthen.

An inverted index maps each term to the list of documents containing it (its postings list), with frequencies and positions. A query reads the postings of its few terms instead of scanning documents. This Java program builds one over four tiny documents and ranks them for a query with a simple length-normalised TF-IDF score: terms that appear in fewer documents are worth more, and a match in a short document counts more than in a long one.

// File: InvertedIndex.javaimport java.util.ArrayList;import java.util.HashMap;import java.util.LinkedHashMap;import java.util.List;import java.util.Locale;import java.util.Map;import java.util.TreeMap;public class InvertedIndex {    public static void main(String[] args) {        Map<String, String> docs = new LinkedHashMap<>();        docs.put("d1", "cheap flights to goa");        docs.put("d2", "goa beach resorts and hotels");        docs.put("d3", "cheap hotels in bengaluru");        docs.put("d4", "flights from bengaluru to goa cheap fares");        // Build: term -> (doc -> how many times the term appears). TreeMap keeps output ordered.        Map<String, Map<String, Integer>> postings = new TreeMap<>();        Map<String, Integer> length = new HashMap<>();        for (Map.Entry<String, String> d : docs.entrySet()) {            String[] terms = d.getValue().split(" ");            length.put(d.getKey(), terms.length);            for (String term : terms) {                postings.computeIfAbsent(term, t -> new TreeMap<>()).merge(d.getKey(), 1, Integer::sum);            }        }        System.out.println("postings for 'goa':  " + postings.get("goa").keySet());        System.out.println("postings for 'cheap': " + postings.get("cheap").keySet());        // Query: score(doc) = sum over query terms of (tf / doc length) x ln(N / df)        String[] query = "cheap goa flights".split(" ");        int n = docs.size();        Map<String, Double> score = new TreeMap<>();        for (String term : query) {            Map<String, Integer> list = postings.getOrDefault(term, Map.of());            double idf = Math.log((double) n / list.size());            for (Map.Entry<String, Integer> p : list.entrySet()) {                double tf = (double) p.getValue() / length.get(p.getKey());                score.merge(p.getKey(), tf * idf, Double::sum);            }        }        List<Map.Entry<String, Double>> ranked = new ArrayList<>(score.entrySet());        ranked.sort((a, b) -> Double.compare(b.getValue(), a.getValue()));        for (Map.Entry<String, Double> r : ranked) {            System.out.printf(Locale.US, "%s  %.4f  %s%n", r.getKey(), r.getValue(), docs.get(r.getKey()));        }    }}

postings for 'goa':  [d1, d2, d4]postings for 'cheap': [d1, d3, d4]d1  0.3171  cheap flights to goad4  0.1812  flights from bengaluru to goa cheap faresd3  0.0719  cheap hotels in bengalurud2  0.0575  goa beach resorts and hotels

Documents d1 and d4 both contain all three query words, but d1 is shorter, so it ranks first. Real engines use refined scoring functions such as BM25 (a standard relevance formula used by default in Lucene-based engines), plus many other signals. Libraries such as Apache Lucene provide inverted indexes, analysis and scoring (Apache Lucene), and search systems such as Elasticsearch and OpenSearch build distributed indexing and querying on top of it (Elastic index basics, OpenSearch introduction).

Index construction is a batch job, naturally expressed as map and reduce steps:

1. Analyse each document: tokenise, lower-case, normalise (and stem where appropriate per language), record positions and fields (title, body, anchor text from other pages' links).
2. Map: emit (term, doc_id, frequency, positions).
3. Shuffle and reduce: group by term (within each index shard), sort postings by doc_id, compress them (postings compress very well using differences between ids).
4. Write immutable index segments per shard, plus per-document scoring data.
5. Publish new segments to serving replicas with versioning, and switch atomically (as with the autocomplete snapshots in article 36).

Choice: document partitioning. With 3 TB of index and, say, 30 GB per shard (kept mostly in memory or on fast storage for latency), about 100 shards. Each shard has several replicas (for example 3 to 5), both for availability and to spread the 3,500 queries per second: every query hits one replica of every shard, so each shard group receives all 3,500 queries per second, divided among its replicas.

Client --> Query front end:   1. results cache (popular queries) -> hit: return   2. parse and analyse the query (same analysis as indexing)   3. scatter to one replica of each of the 100 shards, with a deadline (for example 150 ms)        each shard: read postings, score candidates, return its top 50 (doc_id, score)   4. gather and merge into a global top few hundred   5. second-stage ranking on those candidates with richer signals   6. fetch titles, URLs and snippets for the top 10 from the document store   7. cache and return

With 100 shards, if each shard has a 1% chance of being slow, about 63% of queries would hit at least one slow shard (article 02's 1 - 0.99^100). Mitigations:

• Deadlines and partial results: if a few shards miss the deadline, return results from the rest and mark them partial. Losing 1% of the corpus for one query is better than a slow response.
• Hedged requests: if a shard replica has not answered within its typical p95, send the same request to another replica and use whichever answers first.
• Early termination: order each postings list by a static quality score (or keep a high-quality tier), so a shard can stop after finding enough strong candidates instead of scoring every matching document.
• Result caching: query popularity is highly skewed; caching results for popular queries for minutes removes a large share of load.
• Keep shards in memory or on fast local storage, and keep replicas evenly loaded.

Ranking is done in stages, because expensive signals cannot be computed for millions of matching documents:

1. Per-shard retrieval and scoring: textual relevance (BM25-like scores), field weights (a match in the title or in anchor text from other pages counts more), and term proximity, over all matching documents, with early termination.
2. Global second stage over a few hundred candidates: link-based authority (the idea popularised by PageRank, described in the 1998 paper: a page linked from many important pages is likely important), freshness for time-sensitive queries, page quality and spam signals, language and region match, and diversity (not ten results from one site).
3. Learned ranking models trained on human relevance judgements and interaction signals are common in practice; they are beyond this design's scope.

Spam is an adversarial problem: link farms, keyword stuffing and cloaking (showing crawlers different content) are countered with link-graph analysis, content-quality signals and detection of crawler-specific content.

• A base index, rebuilt in large batches (days to weeks), covers the whole corpus.
• A fresh index, small and updated continuously (minutes), covers recently crawled pages from frequently changing sites; its documents are merged into the base index at the next rebuild.
• Queries search both, and merge results (newer versions of a document override older ones).
• The crawler's priority favours sites that change often and pages that are popular.

This tiering is similar in spirit to how log-structured storage keeps small recent segments and merges them into large ones.

CRAWL PIPELINE (continuous)Frontier (partitioned by host) -> Fetchers (robots.txt, per-host limits) -> raw pages in object storage   -> Parser -> Dedup -> links back to the frontier ; documents -> indexing pipelineINDEX PIPELINE (batch + incremental)Documents -> analysis -> map/shuffle/reduce by shard -> immutable segments -> publish (versioned)Link graph -> authority scores (batch) ; fresh documents -> fresh index (minutes)SERVINGQuery front end -> results cache -> scatter (100 shards x replicas, base + fresh) -> merge   -> second-stage ranking -> document store (snippets) -> response

Search is eventually consistent by nature: the index reflects the web as it was when pages were crawled. Different replicas may briefly serve different index versions during a rollout (minimised by versioned, coordinated switches). What must be prompt is removal: pages subject to legal removal or site-owner requests are filtered at serving time through a blocklist that applies immediately, and removed from the index at the next build.

• Crawl ethically: obey robots.txt (RFC 9309), identify the crawler with a clear user agent and contact information, respect rate limits and removal requests.
• Treat fetched content as hostile: parse in sandboxed processes with limits on size and time; never execute page scripts with access to internal systems; renderers that execute JavaScript run fully isolated.
• Search spam and manipulation: spam detection, link-graph analysis, demotion of low-quality content.
• Query privacy: query logs reveal sensitive information; minimise retention, aggregate, restrict access.
• Abuse of the search API: rate limits, bot detection, protection against scraping of results.
• Safe results: filters for illegal content and optional filtering of explicit content.

"I'd split this into crawling, indexing and serving. For about a billion pages and 100 million queries a day: around 100 TB of raw HTML, a 3 TB inverted index, and 3,500 queries a second at peak.

The crawler keeps a URL frontier partitioned by host, so each host's robots.txt rules and rate limits are enforced in one place; fetchers store raw pages in object storage, and a parser extracts text and links, normalises URLs and removes duplicates. Important and fast-changing pages are revisited more often.

Indexing is a batch pipeline that analyses documents and builds an inverted index, terms to postings lists, as immutable, versioned segments, partitioned by document into about 100 shards with several replicas each. A small fresh index updated every few minutes covers recently crawled pages.

A query checks the results cache, then fans out to one replica of every shard, base and fresh, with a deadline; each shard returns its top candidates by textual relevance, the front end merges them, re-ranks a few hundred with richer signals like link authority and freshness, and fetches snippets. Deadlines with partial results, hedged requests and early termination keep the 100-shard fan-out from blowing up tail latency. The crawler is polite by design, fetched content is treated as hostile, and removals are enforced at serving time."

Text: 5 x 10^9 x 8 KB = 40 TBIndex: 30% -> 12 TBShards: 12 TB / 40 GB = 300 shards

Probability all 200 shards are fast = 0.995^200 = about 0.367Probability at least one is slow = about 63%

• Next: System Design Interview Practice Plan and Job-Ready Checklist, the final article, which turns everything into a plan.
• Related: article 36 (autocomplete) and article 16 (sharding).