Design a Distributed Search System

The ask

A person types a few words into a search box.

The system finds the pages that match the words.

The best matches come back first.

The stored pages are far too many for one machine.

The index is therefore split across many machines.

Estimate the scale

The index holds 50,000,000,000 pages.

That count is the crawler's page count.

One index entry takes 1 KB.

50,000,000,000 entries x 1 KB = 50 TB of index.

The index is split into 100 shards.

50 TB / 100 shards = 500 GB per shard.

Searches reach 5,000,000,000 a day.

5,000,000,000 searches / 86,400 seconds = 57,870 searches per second.

API and data

A search request carries the words the person typed.

The request also carries which page of results the person wants.

A response carries a short list of matching pages.

Each result carries a title and an address.

Each result also carries a short piece of the page text.

The stored data is the index plus the page text for those pieces.

High-level design

A front server takes each search.

The front server sends the search to every shard.

Each shard searches only its own part of the index.

Each shard returns its best matches.

Each match carries a score from its shard.

The front server merges the lists and orders them by score.

The top results go back to the person.

Building the index

An inverted index is a map from each word to the list of pages that contain the word.

Building starts from the stored page text.

Each page is split into words.

Each word is added to the map with the page name beside it.

A common word has a very long list of pages.

A rare word has a short list.

Each shard builds its own inverted index for its own pages.

Answering a search

The front server sends the typed words to every shard.

A shard finds the pages that contain the typed words.

Pages with all the typed words score highest.

The shard scores those pages and keeps its best ones.

Scores favour pages where the words appear often.

Scores also favour pages where the words sit close together.

The front server merges 100 short lists into one list.

Deep dive: fan-out to every shard

Every search touches every shard.

57,870 searches per second x 100 shards = 5,787,000 shard lookups per second across the fleet.

The slowest shard sets the speed of the whole answer.

Each shard keeps its index in memory where it can.

A read of 1 MB from memory takes about 0.25 milliseconds.

Large shards are split so no shard is far bigger than the others.

Copies of a busy shard share its searches.

Freshness and failures

New pages enter the index in small batches.

A changed page replaces its old index entries.

Fresh pages can therefore lag a little behind the crawl.

Each shard has copies on other machines.

A failed shard is served by one of its copies.

A search still answers while one shard copy is down.

Evaluation and trade-offs

Fan-out buys speed. Many machines search at the same time.

Fan-out also costs work. Every search pays for 100 shard lookups.

More shards make each shard smaller and faster.

More shards also make each search touch more machines.

Batch updates keep building cheap but leave fresh pages waiting.

The design accepts that lag for a fast and steady answer.

Scale at a glance