Design a distributed search engine like Elasticsearch

Shards, replicas, scatter-gather and relevance over user documents.

Problem 016 of 100Search and indexingL51 Nov 2026

Concepts this design leans on

Inverted indexes — From every word to the documents that contain it: how search answers in milliseconds.
Sharding and partitioning — Splitting data by key so each machine holds a slice it can handle.
Replication — Keeping copies so reads scale and a dead machine loses nothing.
Scatter-gather — Asking every shard and merging the answers, and the tail latency it brings.
LSM trees and B-trees — The two ways disks store sorted data, and why writes and reads favour different ones.

01The design, in the six-step framework

1. Requirements

Functional

Index JSON documents into named indexes; full-text and structured queries (match, phrase, range, boolean) with relevance scoring, sorting and paging.
Near-real-time: a document is searchable about a second after it is indexed; updates and deletes by id.
Aggregations (counts by field, histograms) over query results.
Out of scope: web-scale crawling, joins across indexes, cross-cluster search.

Non-functional

Clusters from 3 to 1,000 nodes; indexes from gigabytes to hundreds of terabytes (logs are the common large case).
Query p99 under 500 ms for typical queries over a hot index; indexing throughput of hundreds of thousands of documents a second per cluster.
No data loss with one node down; the cluster rebalances itself when nodes join or leave.
Operators can tune the memory versus speed trade-off per index.

Back of the envelope

A log index of 10 TB a day, kept 30 days: 300 TB primary, 600 TB with one replica. At 50 GB per shard (the size at which a single shard still merges and recovers in reasonable time) that is 200 shards per day, 6,000 shards in the cluster, which is why daily indexes and a rollover policy exist.

Indexing 10 TB/day at 1 KB per document = 10¹⁰ docs/day ≈ 116,000 docs/s; one node indexes about 10,000 docs/s per core-set, so 12 to 20 nodes are busy with ingest alone.

Query over 30 days touches 6,000 shards: scatter-gather at that width needs tiering (hot, warm, cold) and index-level routing by date to avoid asking every shard.

2. Core entities

Index: name, mappings (field types and analysers), number of primary shards, replica count, lifecycle policy.
Shard: an index partition; a primary and its replicas; holds segments.
Segment: an immutable mini-index (inverted index, stored fields, doc values) created by a refresh.
Node: roles (master-eligible, data, coordinating), disk and memory, shards hosted.
Cluster state: index metadata and shard routing table, versioned, held by the elected master.

3. API

PUT /{index}/_doc/{id} → index or replace; POST /_bulk → many operations in one request (the only way to reach high throughput).
GET /{index}/_search body: query, sort, size, from or search_after cursor, aggregations → hits with scores, aggregations.
DELETE /{index}/_doc/{id}; PUT /{index} with settings and mappings; POST /{index}/_rollover.

4. High-level design

Indexing then searching:

A bulk request lands on a coordinating node, which hashes each document id to a primary shard and forwards. The primary writes to its translog (durable), adds the document to an in-memory buffer, and sends the operation to its replicas, acknowledging once they have it.
Every second, a refresh turns the buffer into a new searchable segment; segments are merged in the background.
A search goes to one copy (primary or replica) of every shard of the targeted indexes. Each shard returns the ids and scores of its top hits (the query phase); the coordinator merges them, then fetches the full documents only for the final page (the fetch phase).

5. Deep dives

Near-real-time: why one second and not zero?

An inverted index is cheap to query only when it is immutable: sorted posting lists, compressed, with skip structures. So writes never modify a searchable segment; they accumulate in a buffer and become a new segment on refresh. The refresh interval is the trade: one second gives near-real-time search, while a log workload that only queries old data can refresh every 30 seconds and index far faster, because fewer tiny segments are created and merged. Durability is separate from visibility: the translog is fsynced on every request (or every few hundred milliseconds, by setting), so a crash loses nothing acknowledged even if it was not yet refreshed. Deletes are tombstones in a per-segment bitset, dropped at merge.

Scatter-gather over thousands of shards, and the deep-paging trap

A query over 30 daily indexes with 200 shards each asks 6,000 shards. Two mitigations: routing, so that a query filtered to one day or one tenant (if documents are routed by tenant key) goes only to the shards that can contain matches; and tiering, so recent indexes live on fast nodes with more replicas while old ones sit on big cheap disks and answer slower queries. The deep-paging trap: from = 1,000,000, size = 10 makes every shard return a million hits for the coordinator to merge, so the API offers search_after, a cursor on the sort key, and caps from by default. Aggregations are computed per shard and merged, which is exact for sums and approximate for top-N terms unless each shard returns more than N candidates.

Who decides where shards live, and what happens when a node dies?

A small set of master-eligible nodes elects one master by consensus; it owns the cluster state (index settings, which node holds which shard copy) and publishes versions of it to all nodes. Data nodes never make placement decisions. When a data node drops out (missed pings for 30 seconds), the master promotes an in-sync replica to primary for each affected shard and, after a delay to tolerate a quick restart, allocates new replicas on other nodes, which copy segments from the primary with a throttle. The master is not on the query or indexing path, so a master election pauses cluster changes, not traffic. The classic failure is having two masters after a partition; requiring a majority of master-eligible nodes to elect prevents it.

Relevance and the memory trade-offs an operator controls

Scoring uses term statistics (BM25) per shard; because statistics are per shard, scores differ slightly across shards and a global term-frequency pre-pass is available for precision at a latency cost. Memory is the operator's main dial: doc values (columnar per-field storage for sorting and aggregations) live on disk and are paged in; field data for text sorting lives on the heap and is expensive; the filesystem cache holds hot segments. Mappings decide which fields are indexed, which get doc values, and which analysers run, and a mapping mistake (indexing every field of a log line as full text) is the most common reason a cluster runs out of memory. Say this; it shows operational experience.

Index lifecycle for logs: hot, warm, cold, delete

A daily index starts hot: many primaries on SSD nodes, replicas for query load, a one-second refresh. After a day it is read-only, so it is force-merged to one segment per shard (faster queries, less memory), its replica count can drop, and it moves to warm nodes with large disks. After a week it moves to cold nodes or is snapshotted to object storage and searched from there with higher latency; after 30 days it is deleted as a whole index, which is a metadata operation rather than a scan over documents. Rollover by size keeps shards near the 50 GB sweet spot even when the daily volume swings, and the policy runs automatically so an operator does not babysit thousands of indexes.

6. What a strong answer sounds like

Explain the segment model and why immutability is the source of both the speed and the one-second lag.
Separate durability (translog) from visibility (refresh), and replication (primary to replica) from placement (master).
Do the shard arithmetic for a log workload and introduce daily indexes, rollover and tiers as the consequence.
Describe the two-phase query and the deep-paging trap with its cursor fix.
Expect follow-ups on hot-spotting by routing key, on cross-data-centre replication, and on what "one replica" buys in availability versus capacity.
Common mistake: treating shards as free; thousands of small shards cost more memory and cluster-state churn than a few large ones.

02Where this question was reported

interviewkickstart.com · Senior list; also log search.