Requirements

Functional Requirements:

• User is able to serach web via text query
• Return top k relevant results
• Support pagination via cursor for additional results

Indexing

• Continuously index pages produced by web crawler
• Parse, process and build index of those pages

Non-Functional Requirements:

• Highly Available: 99.9% uptime for search
• Low latency: P99 < 500ms for return search results
• Highly scalable: support billions of documents and tens of thousands of search QPS.
• Eventual conistency: Newly crawled pages should be searchable within few minutes
• Durability: Crawled and indexed pages should survive machine failures

Capacity Estimation

Search

• 10B indexed documents
• 100M searches/day
  • 1.2K qps => peak 10x -> ~15K qps

Raw document Storage:

• 1 page ~ 50kb
• 10 billion => 500 TB

Search result size

Assume ~2 KB/result.

If we return 10 results:

At 15K QPS:

Query fanout

Suppose we have:

Each search query fans out to all 100 shards:

This is an important number because our index infrastructure actually handles ~1.5M shard-level requests/sec, not merely 15K.

Each shard should therefore have multiple replicas.

For example:

The retrieval layer sends each query to one healthy replica per shard, so normal traffic is still approximately:

Index size

Suppose the final compressed index is roughly 20% of raw content size as a rough planning assumption:

Then with 3 replicas:

Crawling/indexing

Suppose we want newly crawled documents searchable within 5 minutes.

If we need to process:

that's:

API Design

Only one main api

Search API

GET /v1/search?q=best+toronto+restaurants&limit=10&cursor=

response:

{

results[] => arrray of document data : title, url, snippet,

nextCursor => opaque cursor

}

Cursor

The cursor is opaque to the client but conceptually contains:

Results are deterministically ordered:

This allows the next request to continue after the previous result.

• queryhash prevents using a cursor with a different query.
• indexVersion can provide a consistent search snapshot for pagination. If strict cross-page consistency isn't required, we can allow newer index versions.

High-Level Design

For our system we have two seperate architect, one handling the search, the other handling crawl and indexing pages

Search System

API Gateway

• Entry point for all serach requests
• Applies rate limit - Applied using token bucket per IP

Load Balancer

• Distributes traffic to multiple instances of services

Query Service

• Handles:
  • Query normalization
  • Tokenization
  • Spell corection
  • Common -term handling
  • Any other query parsing

Query Cache

Caches popular query results.

Cache failure isn't fatal; requests fall through to retrieval.

Retrieval Service

• Acts as a coordinator to send queries to relevant index shards in parallel
• Merges returned results from each shard
• Sends the results to ranking service
• Returns back the top k documents

Distributed Index - Shards

• Index are distributed amongst shards
• Each shard has a local inverted index
• We use document-based sharding, for example:

• This distributes documents relatively evenly.
• Each shard applies scoring to documents -> calculated via BM25/lexical score, to return top K.
• Each shard will have replicas for availability

Ranking Service

• Does expensive ranking of documents to come up with top K results sorted via score
• Does ranking in mulitple steps
• Step 1: Apply additonal ranking if necessary to narrow down the number of documents for expensive ML ranking
  • Ranks via freshness, authority, language, Title match, phrase match
• Step 2: The narrow down list is further ranked via expensive ML model ranking.
• Return back the top k results, eg: top 100.

Crawl/Indexing Architect

Crawler workers:

• Continuously fetches pages from the URL frontier.
• It respects:
  • robots.txt
  • crawl delay
  • per-host limits
  • retries/backoff

• Adds additonal checks to pages before sending to kafka for crwaling/indexing pages:
  • Skips pages that are crawled but have no new changes

Kafka

• Acts as durable message queue to hangle async crawling of hundreds of pages simultenously

Indexing pipeline:

• Consumes web crawl events from Kafka
• Handles the actual crawling and indexing
• Steps/Strategy:
  • Store raw html to object storage
  • Parse/normalize - extract title, body, metadata etc
  • Store document metdata in kv store
  • Build index segment - batch few hundred documetns together
  • Publish to index shard

Database Design

We dont have traditional relational database for this system, rather we have:

Object storage like S3 -> Document storage

• Store doc id and raw html content

KV store like Cassandra -> Document crawl metdata

• Used for crawl and indexing management,
• Eg: if content hash not changed, no need to index/crwal again etc

Distributed inverted index

• Primary serving data store
• Stores documents grouped by words eg: term -> posting list

• Also doesnt just store doc ids, but can also store toehr info such as, frequency, position, etc, that can be useful for BM25/lexical scoring

Detailed Component Design

Now we will deep dive into the following topics:

Distributed inverted index - Sharding

• Our system supports 10B documents,
• We cannot store all of those 10B documents in one machine, so we use sharding
• I would prefer using document-based sharding as it gives balancd storage across all shards.

How would querying work when using multiple shards

• Suppose user searched "best toronto restuarants"
• The retrieval service would send query to shards in parallel
• Each shard would search its local index and return top k results

Replication

• Each shard would also have replicas, so if one replica fail, the other replicas can take over serving traffic

Index segments

• Within each shard, i would also use immutable index segments.
• Why?
  • There might be thousands of new/updated documents coming in, so for each modifying the index directly isnt a good idea
  • Rather each shard has segments and new/updated documents will be part of those segments
  • And periodically those segments are merged together in the background - so our segments dont grow huge

If use segments then what about updated document?

• Suppose: Doc123: "Toronto restaurant" -> Part of segment A
• Updated: Doc123: "Best toronto restaurant"
• This update is part of new segment maybe Segment D
• Once segment D is published => Segment A have old vesrion, Segment D has new vesrion
• Search system knows that segment D superseds Segment A version, so serached uses the new version
• Later when segmens are merged the old doc123 from A is discared

Ranking

• Ranking the documents searched via index is the most critical part of the system
• Ranking is done in multiple stages, from cheap ranking frist to expensive ranking last

BM25/Lexical Scoring

• When each shard is asked to serach for documetns for a query.
• Shards just dont return all documetn taht matches the query, the pipeline is:

• BM25/Lexical scoring considers things like;
  • Does documetn contain the query term
  • How long is the document
  • How rare is the term across the entire index
  • How frequently does it occur
  • What poistion are the term in
• Thus applying this gies svore, and top scored documetns are returned

Additional ranking:

• Once shards have returned items and the retrieval service has merged them, the ranking service handles the next, more expensive ranking
• It uses signal such as:
  • Authority
  • Freshness
  • Title match
  • Phrase match
  • Langauge etc

ML Ranking

• By this point, we would have a relatively small set
• So expnsive ranking is affordable
• The list of documetns is ranked by ML ranked producing a relevance score and finally returns back Top k documents

Failure Handling

• Shard fails -> fallback to replicas, so traffic keeps moving
• Indexing pipeline fails -> kafka retains the evetns, twhen the pipeline is back up, the missed evetns are replayed
• ranking service fails -> Fallback to a simpler lexical ranking path such as BM25 if the product's availability requirements prioritize returning results over sophisticated ranking.

Document sharding vs term-based sharding

• Document based - we have each shard with balanced storage
  • Pros:
    • Each shard is balanced storage
    • Easier independent indexing
  • Cons
    • Query fans out ot many shards
• Term based - here we have each shard hadnling specifc terms
  • This makes easy for searching - dont have to query many shards
  • But it introduces hot shard problem -> eg: words like "the" could be aprt of all 10B billion documetns, so shard handling it would become hot.

Why multi-stage ranking i.e, cheap first, expensive later?

• We need fast results -> if all docs are just send to do expsneive ranking, then it might become very heavy,
• Because a search result can produce millions to billions of documents, if we were to do expensive ranking on all docs, we would not adhere to low latency requirment for searches
• Doing cheap ranking first ensures that our expesnive ranking have smaller set to deal with.