The shards were the right count and the wrong size.

00The one-pager

Everything below in a single screen.

01The high-level design

Start here. Everything after this is a zoom into one box.

Elasticsearch is not the source of truth — the operational stores are. The index holds a denormalised copy shaped entirely for retrieval: every field the query filters on, scores on, or returns, and nothing else. That is what removes runtime joins from the hot path.

02The two-minute answer

03Ownership — read this before the interview

A large Elasticsearch cluster is touched by application engineers, platform engineers and infra. Claiming all of it is both untrue and easy to catch — one follow-up about JVM flags or node provisioning finds the edge.

04Context & the storage model

Everything downstream depends on getting this estimate right, and it is the step most people skip — they reason about document count because it is the number they happen to know.

500M candidate documents
average raw JSON              ≈ 2 KB per document
raw JSON size                 ≈ 1 TB
Elasticsearch primary store   ≈ 2 TB     ← _source + inverted index + doc values
                                            + term dictionaries + segments + metadata
with 1 replica                ≈ 4 TB total cluster storage

That ~2× expansion from raw JSON to primary store is the part worth internalising. Elasticsearch is not storing your documents — it is storing your documents plus everything that makes them searchable. Sizing from raw payload underestimates by roughly half.

The existing setup

node type        32 GB RAM data nodes
JVM heap         ~16 GB
OS / off-heap    ~16 GB filesystem cache
index            ~2 TB primary, 5 primary shards
replica          1
cluster total    ~4 TB

→ 2 TB / 5 shards  ≈ 400 GB per primary shard
→ 500M / 5 shards  ≈ 100M documents per primary shard

05Why it was slow

A shard is a Lucene index. The coordinating node fans a query out to the relevant shard copies, each executes locally, and results merge. With few very large shards, every one of those local searches is doing a lot of work.

06Decision 1 — shard sizing

Elastic guidance
  target shard size    10–50 GB
  target document count  below ~200M per shard

we were
  100M docs/shard   ✓  inside the count guidance
  400 GB/shard      ✗  ~8× over the size guidance

pick a middle target for a read-heavy workload
  30 GB per shard

derive
  2 TB / 30 GB  ≈  67 primary shards
  × 1 replica   →  120–140 total shard copies

Smaller shards buy three things at once: lighter per-shard search work, more shard-level parallelism, and dramatically faster recovery and relocation when a node is lost. The third one is invisible until an incident, and then it is the one that matters.

There is an upper bound too — more shards means more fanout per query, more cluster state, and more segment overhead. 67 shards across 8–10 nodes lands at 13–17 shard copies per node, which is a reasonable place to start measuring rather than a number to defend to the death.

07Decision 2 — node capacity

32 GB RAM node
  JVM heap        ~16 GB
  OS cache        ~16 GB      ← what read-heavy search actually runs on

1:24 RAM-to-disk planning ratio
  32 GB × 24    ≈ 768 GB disk per node
  × 70–75% safe ≈ 540–576 GB usable per node

4 TB total / ~550 GB per node  ≈  8 data nodes

practical topology
  8–10 data nodes · 32 GB RAM
  60–70 primary shards · 1 replica
  120–140 shard copies → 13–17 per node

The 32 GB node size is not arbitrary. Elasticsearch leans heavily on the OS filesystem cache for search, and heap is capped near 32 GB because beyond that the JVM loses compressed object pointers. So the sensible shape is ~half heap, ~half cache — and when you need more capacity you add nodes rather than growing them.

08What “1 replica” actually means

Worth its own section because it is the most commonly misread number in a topology, and getting it wrong in an interview is an easy own goal. number_of_replicas: 1 means one replica copy per primary shard — not one replica node, and not one replica of the index.

67 primary shards
+ 67 replica shards          ← one per primary
= 134 total shard copies

placement rule: a replica never sits on the same node as its own primary
                (otherwise losing that node loses the data)

data per node
  8 data nodes  → 16–17 copies × ~30 GB ≈ 480–510 GB
  10 data nodes → 13–14 copies × ~30 GB ≈ 390–420 GB
  both inside the ~540–576 GB safely usable per node

So how many nodes is the cluster, actually?

09Decision 3 — a new index, not an in-place change

Primary shard count is fixed for the life of an index. That single constraint dictates the entire rollout: this is not a settings change, it is a new index, a backfill, a validation pass and a cutover.

PUT candidates_v2
{
  "settings": {
    "number_of_shards":   67,
    "number_of_replicas": 1
  },
  "mappings": {
    "properties": {
      "candidate_id":     { "type": "keyword" },
      "city_id":          { "type": "integer" },
      "job_profile_ids":  { "type": "keyword" },
      "experience_years": { "type": "short"   },
      "expected_salary":  { "type": "integer" },
      "last_active_at":   { "type": "date"    },
      "skills_text":      { "type": "text"    },
      "profile_title":    { "type": "text"    }
    }
  }
}

Then reindex the existing candidate data into it and replay production-like matching queries against both. The fact that you must build a second index turns out to be an advantage: you get a free A/B comparison on identical data before anything is at risk.

10Decision 4 — mapping for read-heavy search

exact filter fields        → keyword / numeric / date
text relevance fields      → text
unused fields              → do not index
large, rarely used fields  → keep out of the _source response

candidate_id               → keyword
city_id                    → integer / keyword
job_profile_ids            → keyword
experience_years           → short / integer
salary_min / salary_max    → integer
last_active_at             → date
skills_text                → text
profile_summary            → text

11Decision 5 — filter first, score second

{
  "query": {
    "bool": {
      "filter": [
        { "terms": { "city_id": [10, 11] }},
        { "terms": { "job_profile_ids": ["field_sales", "retail_sales"] }},
        { "range": { "experience_years": { "gte": 1, "lte": 3 }}},
        { "term":  { "active": true }}
      ],
      "should": [
        { "term":  { "primary_job_profile_id": { "value": "field_sales", "boost": 5 }}},
        { "match": { "skills_text": { "query": "sales customer handling", "boost": 2 }}}
      ],
      "minimum_should_match": 1
    }
  },
  "size": 100
}

Source filtering is the other half. Returning a full candidate profile when the API needs an ID and four fields costs fetch-phase work, serialisation, network and heap — multiplied by 100 hits, multiplied by every query. It is among the cheapest changes available and the most commonly skipped.

12Decision 6 — coordinating-only nodes

Before, application traffic went straight to data nodes, which meant the same machines were doing four jobs: coordinating query fanout and merge, executing shard-level Lucene search, storing shard data, and absorbing indexing load. For a read-heavy workload that is one job too many.

A coordinating-only node stores no shard data — in Elasticsearch, a node with an empty node.roles list. It receives the query, fans it out to shard copies, gathers and merges the results, and returns the response. The effect usually shows up at p99 first, because merge work is exactly what was competing with shard execution under load.

13The experiment

Nothing here was changed blind in production. The sequence was the same every time.

• Capture the baseline. p95/p99 latency, CPU per node, heap usage, GC pressure, search thread-pool queue and rejections, disk usage, shard sizes, slow query logs.
• Create the new index. ~60–70 primaries, replica 1, optimised mappings, source filtering.
• Reindex the candidate data from old to new.
• Replay production-like read traffic. Job-to-candidate search, candidate filters, profile-based scoring, top-N retrieval.
• Compare old vs new on latency, CPU, heap, disk, query throughput and recovery behaviour.
• Cut over safely. Switch the alias or application config; keep the old index available for rollback.

14Alias cutover

POST /_aliases
{
  "actions": [
    { "remove": { "index": "candidates_v1", "alias": "candidates_current" }},
    { "add":    { "index": "candidates_v2", "alias": "candidates_current" }}
  ]
}

The step that gets skipped and should not: result parity. Latency parity is easy to check. A mapping change that quietly alters what matches — an analysed field that should have been a keyword, a boost that reorders the top 10 — produces no error and moves no graph. Shadow queries comparing document counts and top-N quality are the only defence.

15How it got cheaper

The tuning knobs, specifically

Once latency and node pressure improved, cost came down by:

• Right-sizing node count and node type.
• Avoiding oversized machines and unnecessary replicas.
• Removing over-indexed fields, and with them disk and index bloat.
• Reducing the search CPU each query required in the first place.

16The comparison brief

Four dimensions, before and after. If you only remember one screen of this page, make it this one.

The cluster, stated precisely

before                              after
  5 primary shards                    ~60–70 primary shards
  ~400 GB per shard                   ~30 GB per shard
  100M docs per shard                 ~7–8M docs per shard
  data nodes do everything            8–10 data nodes (search only)
                                      + 3 master-eligible
                                      + 2–3 coordinating-only
                                      = 13–16 nodes total

17Anti-patterns avoided

• Sizing shards by document count alone — the mistake that hid this problem for as long as it did.
• Changing shard count in place — not possible; it requires a reindex and alias strategy.
• Scoring before filtering — the single largest source of wasted query CPU.
• Returning the full _source — fetch, serialise and transfer cost for data the API discards.
• High-cardinality dynamic fields in the mapping — unbounded growth on an index you cannot easily change.
• Script scoring where a precomputed field would do — per-document work on every query.
• Deep pagination with from/size — cost grows with offset; search_after exists.
• Treating replicas as HA only — they are a storage and write-cost decision too.
• Scaling the cluster down before reducing query cost — that is how you turn a cost exercise into an outage.
• Tuning heap and CPU before reading the slow logs — optimising what you can see, not what is slow.

18Alternatives, argued fairly

• Throw hardware at it. Fastest fix, and sometimes correct. Rejected because profiling showed the work itself was wasteful, and scaling waste just costs more. The order matters: reduce work, verify, then resize.
• Bigger nodes instead of more. Tempting, but heap is capped near 32 GB for compressed object pointers, and past that extra RAM only helps filesystem cache. More nodes also buys parallelism and faster recovery, which a bigger box does not.
• Route by city or job profile. Would remove fanout. Rejected because a candidate can hold several job profiles and the largest metros would become hot shards — you would trade a uniform cost for a skewed one.
• A vector / semantic search layer. Better fuzzy relevance. Rejected for this problem because the dominant constraints — city, active, experience band, profile eligibility — are hard filters. Semantic ranking would sit after the filter step, and the filter step was the bottleneck.
• Precompute candidate lists per job. Fastest possible read, rejected on cardinality and freshness. The partial version — caching top-N for hot jobs — is worth it, and that is what the cache in the architecture is for.
• Hot/warm tiering by candidate activity. Real money, since inactive candidates are rarely searched. Sequenced after the changes with no correctness risk, because “inactive” needed a definition product had to own — getting it wrong silently excludes people from results.

19Question drill

Read the question, answer it out loud, then open the card and compare.

Sizing

Mechanism

Rollout and cost

20What not to say

21Gaps to fill before the interview

• Ownership, per area. Implemented / recommended / reviewed, for each of the six decisions.
• Real shard numbers. Original and final primary count, and the actual average shard size in GB.
• Real node counts. Data, master and coordinating nodes before and after, and the instance type.
• Real latency. Measured p95 and p99 before and after, by query type.
• The specific clauses that moved from scoring to filter context.
• Cost composition. Whether the ~50% came from node count, instance type, replicas, storage — or a mix.
• Mentorship story. One engineer from the five: gap, coaching, what they owned afterwards.