Distributed Search

Elasticsearch-Like Search Engine

Design the cluster behind a search engine like Elasticsearch: master election, replication and failover, recovery, distributed aggregations, and shard sizing.

Overview

A search cluster spreads one logical index over many machines. Each piece, a shard, is a complete inverted index over part of the documents, with a primary copy that takes writes and replica copies that take reads and stand by. The Search System Design topic explains what happens inside a shard (analysis, BM25, refresh) and how one query is split into a query phase and a fetch phase. This problem is about the distributed system around the shards.

That system has to answer four questions. Who decides where each shard lives, and how do we stop two nodes from both deciding? How does a write reach every copy, and which copies can be trusted after a crash? What happens in the minutes after a node dies? And how do queries and aggregations over hundreds of shards stay fast and honest about what they could not see?

Elasticsearch is the running example because it documents all of these mechanisms, and interviewers usually expect you to know it. The same ideas apply to OpenSearch, Solr and in-house engines.

Questions to ask first

What is searched? — JSON documents: product listings and application logs, 10 billion in total, about 2 KB each
How fast do writes arrive? — 20,000 documents a second on average, 60,000 at peak
How many searches? — 2,000 a second on average, 6,000 at peak; most from one tenant at a time
How fresh must results be? — Searchable within a few seconds
Can we lose an acknowledged write? — No
Is an approximate count acceptable? — For dashboards, yes, if the response says so

What you will be able to do

Explain node roles, the cluster state, and how a quorum of master-eligible nodes prevents split brain.
Trace a write through primary-backup replication, and say when it is durable and when it is acknowledged.
Use sequence numbers, primary terms and the global checkpoint to explain safe failover and fast recovery.
Explain why a distributed terms aggregation can be wrong, and how to bound or fix the error.
Size shards, nodes and data tiers from a stated workload.

Key concepts

Cluster state
Master-eligible quorum
Primary-backup replication
In-sync copies
Sequence numbers and primary terms
Global checkpoint
Translog durability
Delayed allocation
Approximate terms aggregation
Data tiers and rollover

Functional requirements

Create an index with a chosen number of primary shards and replicas
Index, update and delete documents, singly and in bulk
Full-text search with filters, sorting and pagination
Aggregations: counts by term, histograms over time
Report cluster and index health (green, yellow, red)
Snapshot indices to object storage and restore them

Non-functional requirements

An acknowledged write is never lost while at least one in-sync copy survives
Search p99 under about 300 ms for a single-tenant query (a target we chose)
Documents searchable within about 2 seconds of the write (a target we chose)
Losing any one node, or one zone of three, loses no data and keeps search available
A search that could not reach every shard says so in its response

Interview questions

How does the cluster avoid split brain?
Walk me through a write. When is it safe to acknowledge?
The primary dies mid-write. What happens?
Why doesn't the cluster immediately re-replicate when a node disappears?
A dashboard shows the top 10 sellers. Can it be wrong?
How many shards would you create for this workload?
How do you handle a shard that is slow or down during a search?
Replicas already give us copies. Why take snapshots?