Sharding a search cluster

The problem: 600 million links in one index

Snip's search index covers every link: 600 million documents, about 600 GB of index. One machine can't hold it in memory, and can't serve the agencies' query rate alone. The index has to be split.

Split by document, not by term

There are two ways to cut an inverted index:

• By term: one machine owns terms A–F, another G–M. A query touches only the machines owning its terms, but multi-word queries cross machines to intersect, and popular terms make hot machines.
• By document: each shard holds all terms for a subset of documents, a complete small index. A query goes to every shard; each finds and scores its own matches.

Search engines almost always split by document: each shard is independent, and adding documents or shards is simple. In Elasticsearch and OpenSearch each shard is a complete Lucene index.

Scatter, gather, then fetch

A query against a document-sharded index runs in two phases:

1. Query: the coordinating node sends the query to one copy of every shard. Each returns the IDs and scores of its own top 10.
2. Fetch: the coordinator merges the lists into the global top 10, then fetches only those 10 documents from the shards that hold them.

Asking each shard for its top 10 is enough for the global top 10, since every document in the global top 10 is in its own shard's top 10. But page 50 (results 491–500) needs each shard's top 500, so deep pagination gets expensive fast. Use search-after cursors, as in keyset pagination (05.07 Keyset pagination), and cap how deep users can go.

The slowest shard sets the pace

A query fanned out to 20 shards waits for all 20 (03.12 Tail latency and why p99 matters). If each shard is slow 1% of the time, about 18% of queries hit at least one slow shard. Defences: keep shards evenly sized, keep each shard's hot data in memory, use replicas so a slow copy can be avoided, and route around busy nodes. Elasticsearch does the last by default with adaptive replica selection, which picks a copy using the response time of earlier requests and the size of each node's search queue.

Route a tenant's searches to one shard

Snip's searches are always within one owner's links. Fanning every query out to all 20 shards wastes 19 of them. With custom routing, each document is indexed with a routing value, here owner_id, which decides its shard, and a search that passes the same routing value runs only on that shard.

Routing by owner is a big win for Snip, with the usual catch: the largest agency has 40 million links, far more than its share. Like any hot key (05.02 Hot partitions and celebrity keys), it gets special handling: very large owners get their own index, sharded across machines, while the long tail of small owners shares routed shards.

Sizing shards

• Not too big. Elastic's guidance is to aim for shards of 10 GB to 50 GB, and fewer than 200 million documents each. Big shards are slow to move and recover.
• Not too many. Each shard is a separate Lucene index with its own memory and CPU overhead, and every unrouted query touches all of them. Thousands of tiny shards waste resources.
• Replicas serve reads and survive node failure. Two replicas triple the storage and roughly triple read capacity.

For Snip's 600 GB: 20 primary shards of about 30 GB, each with one replica, plus separate indexes for the three largest agencies.

Recap

Interview questions