Sharding and Replication

Elasticsearch search is near-real-time, not real-time. Indexing puts the document in an in-memory buffer and the translog; it only becomes searchable when a refresh turns that buffer into a new searchable Lucene segment, by default about once a second.

Elasticsearch hashes the document's routing value, which defaults to its _id, with Murmur3 and reduces that hash modulo the index's routing-shard count to pick exactly one primary shard. Placement is pure arithmetic, computed on any node, never looked up.

In the query phase each shard runs the search locally and returns only document ids plus sort values or scores; the coordinating node merges these into one globally sorted list. The fetch phase then retrieves the _source of just the winning documents.

Every shard is a separate Lucene index with fixed overhead in heap, file handles, merge work and cluster-state metadata, and every search becomes a task per shard that must be scheduled and merged. Those per-shard costs dominate once shards are small, which is why guidance targets tens of gigabytes per shard.

Document routing derives the target shard from a hash of the routing value and the primary shard count, so changing that count would send lookups to the wrong shard. Replicas are full copies of primaries and sit outside that formula, so they can be added or removed at any time.

A refresh opens the in-memory buffer as a new Lucene segment so recent documents become searchable, without fsyncing anything. A flush performs a Lucene commit that fsyncs segments to disk and trims the translog, which is about durability and recovery, not visibility.

Every write stamps the document with the primary's sequence number and the primary term. A client re-sends both as if_seq_no and if_primary_term; the shard applies the write only if they still match the current document, otherwise it returns a 409 version conflict.

Call it with the index, shard number and whether it is the primary; the response gives unassigned_info explaining why the shard became unassigned, and per-node decider decisions where each NO or THROTTLE names the exact rule blocking allocation on that node.

Send a PUT to the index's _settings endpoint with index.number_of_replicas. It is a dynamic setting, so it applies immediately without closing the index; new copies are allocated and recovered while health shows yellow until they are in sync.

In node.roles, master-eligible nodes can be elected to own the cluster state, data nodes hold shards and execute queries on them, ingest nodes run ingest pipelines, and a node with an empty roles list is coordinating-only: it routes requests and merges results.

Both create a new index rather than resizing in place, and both need the source made read-only first. _split needs a target primary count that is a multiple of the source count and compatible with number_of_routing_shards; _shrink needs a target count that divides the source count, green health, and a copy of every shard gathered on one node.

refresh=wait_for holds the response until the next scheduled refresh makes the change visible, forcing no extra work. refresh=true forces an immediate refresh of the affected shards, creating an extra segment per request and much more merge pressure.

Set cluster.routing.allocation.enable to primaries so replicas of the departing node are not rebuilt elsewhere, stop non-essential indexing and flush, restart the node, then set the setting back to all and wait for green. Delayed allocation covers short absences automatically.

Elasticsearch maintains a voting configuration: the set of master-eligible nodes whose votes count. An election needs a strict majority of that set, so the cluster survives losing fewer than half of it. Below that it has no master and rejects cluster-state changes.

By default each shard scores using only its own term statistics, so identical documents can score differently per shard. dfs_query_then_fetch adds a preliminary round trip that gathers term and document frequencies from all participating shards so scoring uses global statistics.

Successive searches may be served by different copies of the same shard, and a primary and its replica hold different segment layouts and different numbers of not-yet-purged deleted documents, so their local statistics differ slightly. A constant preference value pins requests to the same copies.

It caches per-shard search results keyed on the whole JSON request body, and by default only for requests with size:0 — aggregations, suggestions and the total hit count. A now value resolves to a new instant each request, producing a new key and a guaranteed miss.

You cannot change the primary count of existing indices, so fix it in two directions: shrink and force-merge the already-created ones down to one primary, and change the index template plus the rollover trigger so future indices are created with far fewer shards and roll on primary shard size rather than on the calendar.

The coordinating node routes the request to the primary, which validates it, applies it locally with a sequence number and primary term, then forwards it in parallel to every in-sync replica. Once they respond the primary acknowledges; a failed replica is reported to the master and removed from the in-sync set.

With async durability the translog is fsynced on a timer rather than per request, so a node crash or power loss can lose every acknowledged write since the last sync interval. The gain is far fewer fsyncs and higher indexing throughput.

Put one master-eligible node in each zone so a majority survives losing one, spread data nodes evenly, tag every node with a zone attribute, enable allocation awareness on that attribute so shard copies land in different zones, and keep at least one replica per index.

Weigh the fan-out saved against the skew created. Custom routing turns every tenant search into a single-shard search, which multiplies search throughput, but it concentrates each tenant on one shard, so an outsized tenant becomes a hot shard the cluster cannot rebalance away.

Benchmark one shard loaded with representative data and real queries to find the size where latency degrades, then divide the projected data volume by that size. Where growth is genuinely unknown, design so the number never has to be right forever: roll over to new indices sized on shard size.