Designing a Distributed Search Engine: Crawling, Indexing, and Ranking Billions of Pages

Why Study a Distributed Search Engine?

A web search engine has to solve two very different hard problems and stitch them together: continuously discovering and reading a large fraction of the public web (tens of billions of pages), and then answering a query against that entire corpus in under 200ms. Google indexes hundreds of billions of pages and serves on the order of 100,000+ queries per second worldwide. No single machine can hold that index or answer that query volume, which makes this one of the richest "distributed systems" interview problems - it forces you to reason about sharding strategy, scatter-gather query fan-out, and caching all at once.

Requirements Analysis

Functional Requirements

1. Crawling: Discover and download pages across the web, following links
2. Indexing: Build a searchable index from crawled content, updated as new pages appear
3. Query Processing: Accept a text query and return ranked, relevant results
4. Ranking: Order results by relevance (content match + authority signals)
5. Freshness: Re-crawl and re-index pages that change, at a cadence appropriate to how often they change

Non-Functional Requirements

High-Level Architecture

Distributed Crawler Architecture

The URL Frontier

The URL frontier is the queue of URLs waiting to be crawled. It isn't a simple FIFO - it has to balance two competing goals: prioritizing important/fast-changing pages, and never hammering any single domain with concurrent requests.

Politeness Policy

Politeness is a hard constraint, not an optimization: crawling too aggressively can effectively DoS a small site, and most sites will simply block a crawler's IP range if it misbehaves. The standard approach:

1. Per-domain queues: every URL is routed to a queue keyed by its domain, so a crawler worker only ever has a small, bounded number of requests in flight against any one host.
2. Minimum delay between requests to the same domain: even with a per-domain queue, enforce a minimum gap (commonly seconds, sometimes longer) between consecutive requests to the same host.
3. robots.txt compliance: fetched once per domain and cached; disallowed paths are never enqueued, and a Crawl-delay directive (if present) overrides the default delay.
4. Respecting HTTP response signals: back off further on 429/503 responses, treating them as an explicit "slow down" signal from the origin server.

Deduplication via URL Fingerprinting

The same content is reachable through many different URLs (tracking parameters, session IDs, http vs https, trailing slashes). Before enqueueing a discovered URL, the crawler normalizes it (strip fragments, sort query params, lowercase host) and computes a fingerprint (a hash of the normalized URL) to check against an already-seen set. At web scale this seen-set is far too large to hold as an exact hash set in memory, which is exactly the problem a Bloom filter is built for - a compact probabilistic membership test that says "definitely not seen" or "probably seen," letting the crawler skip the vast majority of duplicate URLs with a small, fixed memory footprint and an acceptable false-positive rate. This guide covers the data structure itself in more depth in the Bloom Filters article - the short version here is that it's the standard tool for this exact "have I seen this URL before" check at billions-of-URLs scale.

The Inverted Index

Why Not Scan Every Document?

A naive search ("does this document contain the word 'kubernetes'?") scanned across billions of documents at query time is hopeless at any meaningful latency budget. The inverted index flips the relationship: instead of document -> words it contains, it stores word -> list of documents containing it (a "posting list"). A query for a term becomes a direct lookup into this structure instead of a scan.

Each posting list entry typically stores more than just a document ID - term frequency within that document, positions (for phrase queries), and field information (title vs body) - so the index can support both simple term matching and richer ranking signals without a second pass over the raw document.

This is the same conceptual model used by real systems like Lucene/Elasticsearch, which is why the brief describes this design as "Lucene-style."

Index Sharding Strategy

A single machine cannot hold an inverted index for the entire web, so the index is sharded across many machines. There are two fundamentally different ways to split it, and the choice shapes the entire query path.

In practice, large search engines favor document-based sharding: even though every query touches every shard, each shard does bounded, parallelizable work (find its own local top-K), and the fan-out/merge pattern is far easier to operate and rebalance than trying to keep term-based shards load-balanced as the vocabulary and its skew evolve over time.

Query Processing Pipeline

1. Tokenization: normalize the raw query the same way documents were tokenized at index time (lowercasing, stemming, stopword handling) so terms actually match posting list keys.
2. Query parsing: interpret structure - implicit AND between terms, explicit OR, quoted phrases, filters (site:, filetype:).
3. Scatter: the query coordinator sends the parsed query to every relevant shard (all shards, under document-based sharding) in parallel.
4. Per-shard local top-K: each shard scores only its own documents and returns its best K candidates, not its entire result set - this bounds the data sent back over the network regardless of how many documents actually matched.
5. Gather + global merge: the coordinator merges the per-shard top-K lists into a single global ranking (a K-way merge, since each shard's list already arrives sorted).
6. Rerank (optional second pass): a more expensive ranking model can rerun on just the merged candidate set - now small (hundreds, not billions) - to refine final order before returning results.

Ranking Signals

Ranking a candidate document combines multiple signal families - the exact scoring math is a research topic in its own right, but the conceptual building blocks are:

• Term-frequency-based relevance (TF-IDF / BM25): how often query terms appear in a document, offset by how common those terms are across the whole corpus (so matching a rare term counts more than matching "the"). BM25 is the modern standard, refining raw TF-IDF with saturation (repeating a term 100 times doesn't count 100x as much) and document-length normalization.
• Link-based authority signals (PageRank-style): a document's importance inferred from the link graph - pages linked to by many other important pages rank higher, independent of their own text content. This is what let early Google outperform pure keyword-matching engines.
• Freshness and metadata signals: recency for time-sensitive queries, page quality/spam signals, and (in modern engines) learned ranking models trained on click behavior.

Final ranking is typically a weighted combination of these signal families, often with a fast first-pass score (used for the initial per-shard top-K) followed by a more expensive rerank pass on the smaller merged candidate set - the same two-phase pattern used in the query pipeline's merge step above.

Multi-Layer Caching

Cache invalidation is comparatively gentle here versus other systems: search results tolerate slight staleness far better than, say, an inventory count, so a short TTL (seconds to low minutes) rather than active invalidation is usually sufficient - freshness-critical queries (like breaking news) can be explicitly excluded from caching by the query parser.

Database Schema and Index Metadata

While the inverted index itself is typically a custom on-disk structure (not a relational table), the system still needs relational/metadata stores around it for crawl state, document metadata, and shard routing.

Crawl Queue State

CREATE TABLE crawl_urls (
    url_fingerprint BIGINT PRIMARY KEY, -- hash of normalized URL
    url TEXT NOT NULL,
    domain VARCHAR(255) NOT NULL,
    priority SMALLINT DEFAULT 0,
    last_crawled_at TIMESTAMP,
    next_crawl_at TIMESTAMP,
    status VARCHAR(20) DEFAULT 'pending' -- pending, in_progress, done, failed
);
 
CREATE INDEX idx_crawl_domain ON crawl_urls(domain, next_crawl_at);
CREATE INDEX idx_crawl_priority ON crawl_urls(status, priority DESC, next_crawl_at);

Document Metadata

CREATE TABLE documents (
    doc_id BIGINT PRIMARY KEY,
    url TEXT NOT NULL,
    title VARCHAR(500),
    content_hash BIGINT NOT NULL, -- detects unchanged content on re-crawl
    authority_score FLOAT DEFAULT 0, -- precomputed PageRank-style score
    indexed_at TIMESTAMP DEFAULT NOW(),
    shard_id INT NOT NULL
);
 
CREATE INDEX idx_documents_shard ON documents(shard_id);
CREATE INDEX idx_documents_content_hash ON documents(content_hash);

Shard Routing

CREATE TABLE index_shards (
    shard_id INT PRIMARY KEY,
    doc_id_range_start BIGINT NOT NULL,
    doc_id_range_end BIGINT NOT NULL,
    host VARCHAR(255) NOT NULL,
    status VARCHAR(20) DEFAULT 'active'
);
 
CREATE INDEX idx_shard_range ON index_shards(doc_id_range_start, doc_id_range_end);

Capacity Estimate

A rough back-of-envelope sizing exercise, useful for grounding the design in real numbers during an interview:

These numbers exist to justify design decisions (why sharding is mandatory, why compression matters, why caching hot queries pays off), not to be memorized exactly - interviewers care that you can derive them from first principles.

Scaling Challenges & Solutions

Key Takeaways

1. Separate crawling from query serving as two independent subsystems connected only by the index - they have completely different latency and consistency requirements
2. Politeness is a hard constraint on crawling, not an optimization - per-domain rate limiting and robots.txt compliance are non-negotiable at web scale
3. The inverted index turns search into a lookup, not a scan - this single data structure choice is what makes sub-200ms search over billions of documents possible at all
4. Document-based sharding trades wider fan-out for even load - and is preferred over term-based sharding because query traffic skews heavily toward common terms
5. Caching works because query popularity follows a power law - a small set of popular queries accounts for a disproportionate share of traffic, making even a small cache highly effective

Follow-Up Questions to Consider

1. How would you handle personalized or localized search results without breaking the shared index/cache model?
2. How would you detect and handle spam pages or link farms trying to manipulate authority scores?
3. How would you support autocomplete/query suggestions with sub-50ms latency?
4. How would you re-balance index shards as the corpus grows from 10B to 100B documents without downtime?
5. How would you extend this design to support image or video search, where the "content" isn't plain text?