Searching 730 billion tweets in under 200ms — the inverted index, the shard fan-out, and the architecture that makes a multi-petabyte index feel instant
This deep-dive applies the 4-step HLD interview framework. As you read, map each section to Requirements → Entities → APIs → High-Level Design → Deep Dives, and notice which of the 8 common patterns and key technologies are at play.
Imagine Sarah, sitting on her couch during the World Cup final, types "world cup final" into the Twitter search bar. Within the next 200 milliseconds, Twitter must dig through 730 billion tweets ever posted, find the few hundred thousand that mention all three of those words, rank them by how recent and how popular they are, and return the top 20 to her phone. The data being searched is multi-petabyte; the answer must feel instant. That's the system we're designing.
The naive approach — "scan every tweet looking for the words" — would take hours per query and saturate every database in the cluster. Twitter Search instead pre-builds a special data structure called an inverted index that maps every English word to the list of tweets that contain it. When Sarah searches, we don't search the tweets — we search the index. That single architectural choice is what makes the system possible.
Before drawing a single box, pin down what the system must do. In an interview, asking these questions early shows you're not just reaching for a memorized solution.
Numbers are not optional in HLD. They drive every architectural choice — sharding count, server RAM, fan-out width — so do them out loud, even if rough.
Assume 1.5 billion total users, of which 800 million are daily active. Twitter sees 400 million tweets per day (roughly the public-tweet rate at peak Twitter scale).
~4,600 / sec
400M / 86,400
500M / day
~625 per active user / month
~5,800 QPS
500M / 86,400
300 bytes
text + metadata, no media
400M tweets/day × 300 bytes = 120 GB/day = ~1.4 MB/sec of new tweet data. Over 5 years that's ~200 TB of tweet text.
Apply an 80% capacity model (always keep 20% headroom) plus 2× replication for durability: 200 TB / 0.8 × 2 ≈ 500 TB total. Spread across servers with 4TB SSDs each, we need ~125 storage servers.
The index is separate from the raw tweets. Assume 500K commonly-used English words plus 200K proper nouns/entities — call it ~700K total vocabulary. Each tweet contains roughly 15 indexable words (after dropping stopwords like "the", "is", "of"). For 730B historical tweets, the inverted index is roughly:
700K words × ~5 bytes (word) + 730B tweets × 5 bytes (tweet_id) × 15 words/tweet ≈ ~21 TB index.
To hold this in RAM (so query latency is microseconds, not milliseconds), at 144 GB RAM per server: 21 TB / 144 GB ≈ ~152 index servers.
| Metric | Value | Why it matters |
|---|---|---|
| New tweets / sec | 4,600 | Drives index-update throughput & Kafka sizing |
| Search QPS | 5,800 | Drives aggregator + index-server fleet size |
| Storage growth | 120 GB/day | Forces sharded primary store (Cassandra) |
| 5-yr tweet storage | 500 TB | Sized across ~125 servers |
| Inverted index size | ~21 TB | Sized across ~152 RAM-resident index servers |
| Index lag SLO | < 30 sec | Tweet→searchable latency budget |
One endpoint carries 99% of the search traffic. Defining it early locks the contract before architecture.
REST API surface// Search — read path, the only endpoint users hit directly GET /api/v1/search { "api_key": "abc123...", // rate-limit identity "query": "world cup final", // space-separated terms; AND by default "count": 20, // page size, max 100 "sort": "relevance", // "relevance" | "recency" "page_token": "eyJzaGFyZCI6...", // opaque cursor for pagination "filters": { "lang": "en", "from_date": "2026-05-01", "to_date": "2026-05-07", "user": null, "has_media": false } } → 200 OK { "results": [ { "tweet_id": "1789234...", "text": "Argentina just won the World Cup final!!!", "user": { "id": 42, "handle": "@futbolfan" }, "created_at": "2026-05-07T14:02:31Z", "like_count": 12453, "retweet_count": 4821, "score": 0.94 }, ... ], "next_page_token": "eyJzaGFyZCI6...", "took_ms": 142 }
page_token instead of an integer offset? An offset like page=5 forces every shard to scan the first 100 results just to skip them — wasteful and slow. The token encodes "where each shard left off" — for example "shard-7 stopped at tweet_id 1789234, shard-12 at 1789201". On the next page, each shard resumes from its own cursor in O(log N) time. Same trick Google, Elasticsearch and DynamoDB all use.api_key? A single hot query like "the" (millions of matches) is 1000× more expensive than "giraffe poetry" (a few hundred). Without rate limits, one bot could exhaust the cluster. We rate-limit by API key (e.g., 100 searches/min for free tier) and reject overly broad queries with a 400.The single most important storage insight: we have two completely different storage systems with completely different access patterns. Confusing them is the most common mistake.
The source of truth. Every tweet ever posted lives here, keyed by tweet_id. We embed the timestamp into the tweet_id so the IDs are roughly time-sortable (think: Snowflake-style 64-bit IDs where the upper bits are epoch milliseconds).
Access pattern: "give me the full tweet object for tweet_id X" — pure key-value, no joins, no scans. Sharded by tweet_id range (or hash) across ~125 nodes.
Why Cassandra: handles wide-column key-value at petabyte scale, multi-region replication out of the box, tunable consistency.
A precomputed map of word → [tweet_id, tweet_id, ...]. Built incrementally from the tweet firehose, held in RAM across ~152 servers, never queried for the actual tweet content.
Access pattern: "give me tweet_ids that contain word X" — return a sorted list. Multi-word queries do set intersection across multiple lookups.
Why a separate system: the index is queried 1000× more often than tweets are written, has very different memory/access characteristics, and updating it can be done asynchronously without blocking writes to the primary store.
This is the section that wins or loses the interview. We'll build the architecture in three passes: the simplest thing that could plausibly work, why it falls apart, and the production shape where every box justifies itself. Numbers from §3 drive every decision.
The simplest possible system: store every tweet in a SQL database. To search, run SELECT * FROM tweets WHERE text LIKE '%world cup%'. One server, one DB.
Four concrete failures emerge the moment traffic shows up:
LIKE '%word%' with leading wildcard cannot use any index — the database must read every row. Scanning 730 billion rows on commodity hardware takes several hours per query, while our SLO is 200 milliseconds. This isn't a tuning problem; it's an algorithmic one.
SQL's LIKE is dumb string matching. It won't tokenize ("world-cup" vs "world cup"), won't stem ("running" vs "run"), won't drop stopwords, won't handle hashtags or mentions. A search for "running shoes" misses every tweet that says "runners need new shoes".
SQL returns rows in whatever order the storage engine pleases. We need to rank by recency, popularity, and engagement — none of which a LIKE query can express. We'd have to fetch all matches and sort in app code, which compounds the scan problem.
Even one search at a time would lock the database for hours. At our 5,800 QPS target, the queue would grow infinitely within seconds. Adding read replicas doesn't help — each replica has the same scan problem.
The single most important insight in this design: don't search the data — search a precomputed index of the data. This is the same trick used at the back of every textbook: the index lets you go from "I want to know about photosynthesis" straight to "page 247", without flipping through every page in the book.
Read every tweet, ask "does it contain my word?", keep the matches. Cost: O(N) where N = 730 billion. Hopeless.
Precompute, for every word in the language, the list of tweet IDs that contain it. To search, look up the word and return its list. Cost: O(log V) where V = 700K vocabulary. Effectively constant time.
Here's what the index looks like, conceptually:
For a multi-word query like "world cup final", we look up each word's list and intersect them — only tweets that appear in all three lists match. Modern intersection algorithms on sorted ID lists are extremely fast (linear in the size of the smallest list, with skip pointers shrinking that further).
Two more pieces fall out of this insight:
Now the full picture. Every node is numbered — find its matching card below to see what it does and what would break without it. The architecture splits into three planes: the Query Plane (handles searches), the Index Plane (holds and updates the inverted index), and the Source Plane (stores the raw tweets and the helper data structures).
Use the numbers in the diagram above to find the matching card. Each one answers what is this, why is it here, and what would break without it.
Sarah's mobile app, the desktop web client, or a third-party app calling our public API. The client sends a single GET /api/v1/search with the user's query and waits for a JSON list of ranked tweets. From the client's perspective, the entire 152-server index cluster is one URL.
Solves: nothing on its own — but every design choice flows backward from "what does the client experience?" The 200ms latency budget, the page-token pagination, and the rich result objects all exist because of what the client needs.
The traffic cop. Distributes incoming HTTPS across the Search API server pods, terminates TLS so backend pods don't pay the crypto cost, and yanks unhealthy pods out of rotation via 5-second health checks. Public-facing — typically AWS ALB or nginx behind a global anycast IP.
Solves: single-server bottleneck and single-server failure. Without an LB, one app crash takes down search for everyone. With it, we lose 1/N of capacity for a few seconds until health checks fail the bad pod out.
Stateless service that handles the public /search endpoint. Per request it: parses the query, validates it, checks the cache, on cache miss delegates to the aggregator, calls the ranking service for final scoring, calls the suggestion service for "did you mean", and returns the assembled JSON. Easy to scale because it's stateless — add pods behind the LB.
Solves: isolating user-facing concerns (auth, rate-limits, response shaping) from the heavy machinery of the aggregator and index cluster. Without this tier, every change to the JSON response shape would touch the aggregator — bad coupling.
The heart of the query plane. Takes a tokenized query (e.g. ["world", "cup", "final"]) and fans it out to every index shard in parallel. Each shard returns its local top-K matching tweet_ids; the aggregator merges them, takes the global top-K, fetches the full tweet objects from the Tweet Source DB, and hands them to the ranking service. Think of it as the "scatter-gather" coordinator.
Solves: the fundamental sharding problem. The index is too big for one server (21 TB), so it's split across 152 servers — but a query for "world" might match tweets that live on any of those servers. Without the aggregator, the client would have to query all 152 itself. The aggregator hides the sharding from the rest of the system.
A fleet of ~152 servers, each holding a partition of the inverted index in RAM. Each server indexes only the tweets it owns (we shard the tweets, not the words — see §7) so it sees the full vocabulary but only its slice of tweets. On a query, each server returns the top-K matching tweet_ids it knows about, scored by a local relevance function. Stateless from the network's view but stateful in memory.
Solves: turning a 21 TB index into something that fits in RAM and serves microsecond lookups. Without this cluster, every search would hit disk on a 21 TB SSD array — adding 50-100ms per shard, and the per-shard latency adds up across the fan-out.
The async pipeline that keeps the index fresh. It consumes the tweet firehose from Kafka (every new tweet shows up here within milliseconds of being posted), tokenizes the text (split on spaces, drop punctuation, lowercase), removes stopwords ("the", "and", "is"), applies stemming ("running" → "run"), and pushes the resulting (word, tweet_id) pairs to the appropriate index server. It also writes an entry to the Reverse Index so we can find the home shard of any tweet later.
Solves: the index-freshness problem. We have 4,600 new tweets per second; the index must absorb them within seconds (our SLO) without blocking searches. Without the builder, the index would be a static snapshot — useless for "live" searches like World Cup commentary. The async pipeline lets us scale the builder fleet independently of the index-server fleet.
The source of truth — every tweet ever posted, keyed by tweet_id (with the timestamp embedded so IDs are roughly time-sortable). 500 TB across ~125 nodes, sharded by tweet_id range. The aggregator hits this only after intersection narrows the result set down to a few thousand candidates — never to scan, only to fetch the body for the top results.
Solves: durable storage of the actual tweet content. The index returns IDs; we still need to show Sarah the tweet text, the author, the like count. Without the source DB, the index is a list of pointers to nowhere.
A simple HashMap: tweet_id → which index shard owns this tweet's index entries. Maintained by the Index Builder as a side-effect of indexing. Held in memory by a small replicated service, persisted to disk for restart safety.
Solves: fault recovery — the unsung hero of this design. When an index shard dies and we want to rebuild it, we don't want to re-tokenize 730B tweets. With the reverse index, the recovering shard asks "which tweets did I own?" and pulls only those (a few billion) from Cassandra — cutting recovery from days to hours. Without it, a shard loss is a multi-day rebuild.
An in-memory key-value store holding the final ranked result list for hot queries. Cache key = hash of the (query, sort, filters) tuple; value = the final JSON response with tweet IDs and scores. Eviction: LRU. Replicated for fault tolerance; sharded by query hash for capacity.
Solves: the viral-query problem. When the World Cup final ends, "world cup final" gets queried 10,000 times per second for an hour. Without cache, every one of those queries fans out to all 152 index shards — wasted work since the answer is the same. With cache, only the first query does the work; the next 36 million served from RAM in microseconds.
Scores the candidate tweets by a combination of: recency (newer = higher), popularity (likes + retweets), engagement (replies, click-through), term frequency (how many query words matched), and personalization (does the user follow the author?). Each index shard returns top-K by a fast local score; the ranking service does the final, more expensive ML-based re-ranking on the top few hundred candidates.
Solves: the difference between "results that match" (easy) and "results Sarah actually wants to see" (hard). Without ranking, search would return matches in arbitrary order — and a search for "world cup" would surface a 10-year-old tweet ahead of the goal that just happened.
Powers "did you mean" and the typeahead dropdown as Sarah types. Built from a separate prefix index (trie or finite-state transducer) plus a Levenshtein-distance spell-corrector trained on common search misspellings. Returns suggestions in under 30ms so the typeahead feels instant.
Solves: the "user typed it wrong" problem. Without spell-correct, a search for "wrold cup" returns zero results and a confused user. With it, we silently suggest "world cup" and the search succeeds.
Two real flows, mapped to the numbered components above:
GET /search?query=world+cup+final. The Load Balancer ② routes it to a Search API Server ③.["world", "cup", "final"], sends it to the Aggregator ④.tweet_id = 1789234.["final", "argentina", "deserved", "world", "cup"].1789234 → shard-12 to the Reverse Index ⑧.The inverted index is the soul of this system. Let's open it up: how it's structured, how big it gets, and the two competing ways to shard it.
At its simplest, the inverted index is a HashMap<String, List<Long>>: keys are vocabulary words, values are sorted lists of tweet IDs that contain that word. Think of the index at the back of a textbook — except instead of "page numbers", we store tweet IDs, and instead of one book, we're indexing 730 billion of them.
Production indexes layer on more sophistication: each posting list also stores per-word frequencies (for TF-IDF scoring), positions (for phrase queries like "world cup" as opposed to "world" AND "cup"), and skip pointers (so intersection across two lists can leap forward in O(log N) instead of O(N)). Lucene/Elasticsearch is essentially a hardened, on-disk version of this same idea.
Let's verify our 21 TB number from §3:
At 144 GB RAM per server (a beefy but standard box): 21 TB / 144 GB ≈ 152 servers. Add 2× replication for fault tolerance → ~300 index servers in production.
This is the most interesting trade-off in the design. We have to split 21 TB across 152 servers; the question is how.
Each server owns a slice of the vocabulary. "world" and everything starting with "wo-" lives on shard-7; "cup" lives on shard-3. To answer a query, the aggregator sends one word to each owner.
Pros: a query for one word touches one server. Minimal network fan-out.
Cons: hot words destroy the design. The shard owning "the" (still indexed for some queries) gets every query for English text. Trending words like "world cup" during a final concentrate all load on one server — that server melts while 151 others sit idle. Also, intersecting two words requires shipping one of the lists across the network — for "world" with 300K IDs that's a 1.5 MB transfer per query.
Each server owns a slice of the tweet space (e.g., shard-12 owns tweet_ids 1789000000 – 1789999999). It indexes only its own tweets, so it sees the entire vocabulary but only ~5 billion tweets. Every shard has the word "world" — but each holds only its slice of "world"-containing tweets.
Pros: load is uniformly distributed by construction (tweet IDs are roughly time-ordered and randomly hashed). No hot shards. Intersection happens locally on each shard with no network transfer of large lists. Adding a new shard just changes the hash range — no global rebalance.
Cons: every query fans out to every shard (152-way fan-out). That sounds expensive but each shard does its work in parallel, so wall-clock latency is set by the slowest shard, not the sum.
The aggregator's job in strategy (B) is exactly what makes it work: it gathers the per-shard top-K, merges them into a global top-K. With 100 results per shard and 152 shards, the merge is over 15,200 IDs — trivial in milliseconds.
Index servers hold their state in RAM. RAM is volatile. Servers crash. So what happens when shard-12 dies at 14:02 with 5 billion tweets' worth of index in memory?
Each shard is replicated to a standby server in a different rack. The Index Builder writes to both primary and standby on every update. If the primary fails, the standby takes over within seconds — a brief latency bump (the warm-but-cold standby's RAM caches need a minute to settle), no data loss.
But what if both die — say a power-rail failure takes down a whole rack? Now we have to rebuild the shard's index from scratch.
"Re-tokenize all 730 billion tweets and re-route each to its owning shard." This works, but takes days. During those days, queries that should have hit shard-12 return incomplete results — potentially missing huge swaths of recent content.
Recall: the Reverse Index is a HashMap of tweet_id → owning shard, maintained by the Index Builder as a free side-effect. When shard-12 needs to recover:
Now we fetch only the 5 billion tweets that actually belong on shard-12 instead of all 730 billion. Recovery drops from days to hours. The Reverse Index itself is small (~3.7 TB across the cluster, sharded by tweet_id) and easily replicated — we lose at most a small region of it on a single failure, recoverable from a daily snapshot.
The single biggest performance lever for hot queries. Without a cache, every search of "world cup final" during the actual final does the full 152-way fan-out — even though the answer is identical for every querier in any given second.
The final ranked result — the top-20 tweet IDs and their scores, keyed by hash(query, sort, filters). Not the per-shard intermediate lists; those vary across shards and aren't independently reusable. The whole-result cache is a single Redis GET for any cache hit.
LRU with a short TTL (e.g., 30 seconds). Hot queries stay hot; the TTL ensures the result reflects recent tweets — Sarah doesn't get a 5-minute-old top-20 during a fast-moving event. We can drop TTL to 5 seconds for extreme freshness on trending topics.
~5,800 QPS, ~30s TTL → at most ~175K hot entries cached. Each entry is small (~1 KB of IDs + scores) → ~175 MB total. Trivially fits on a single Redis instance, but we shard across 4-6 nodes for fault tolerance and to spread the QPS.
Load balancers sit at two layers in the search path, each playing a slightly different role.
Public-facing LB (AWS ALB / nginx). Distributes incoming HTTPS across the Search API pods. Health-checks every 5s, evicts unhealthy pods. Terminates TLS so backend pods don't pay the crypto cost. Round-robin works fine here — every request is roughly the same shape from the API's perspective.
Internal LB (envoy or a service-mesh sidecar). Routes API requests to aggregator pods. Switches to least-connections once the cluster is large enough that one slow aggregator (e.g., processing a viral query) shouldn't pile up new requests behind it.
The aggregator is the load balancer for the index cluster — it deliberately sends the query to every shard (since we shard by tweet_id, every shard might have matches). There's nothing to "balance" in the traditional sense; it's a deterministic fan-out. Within a shard's replicas (primary + standby), the aggregator picks the healthiest one, that's it.
"Find the matches" is the easy half of search. "Show Sarah the matches she wants to see" is the hard half. Without ranking, a search for "world cup" might surface a 10-year-old tweet ahead of the goal scored 2 minutes ago.
Production search engines almost universally use a two-stage approach for cost reasons:
Each index server scores its local matches with a cheap formula: score = recency_boost × tf_idf × log(1 + likes). Returns top-K (say, top 100) per shard. Fast, parallel, no cross-shard data needed.
The Ranking Service ⑩ takes the top ~200 candidates from across all shards and runs them through an ML model (gradient-boosted tree or small neural net) that scores on richer features: full engagement signals, the searcher's social graph (do they follow the author?), language quality, freshness, has-media boost. Returns the final top 20.
LIKE '%word%' with a leading wildcard does a full table scan — O(N) over 730B rows = hours. The inverted index pre-pays the cost of "for every word, what tweets contain it?" once, at index-build time. Then every query is a hash-map lookup → O(1) per word, plus a sorted-list intersection that's roughly O(min list size) with skip pointers. The win comes from doing the work once, in batch, instead of repeating it for every query.(word, tweet_id) pairs to the appropriate Index Server. Because the pipeline is async, search reads never block on writes — and we can scale the builder fleet independently (more workers = faster catch-up after a backlog). End-to-end lag is typically under 10 seconds; our SLO is under 30 seconds. If a builder falls behind, Kafka buffers without dropping events.(query, filters) tuple over and over. With a 30-second cache TTL, only the first query in each TTL window does the 152-way fan-out; the next ~300,000 are served from Redis in microseconds. Hit rate during a viral spike commonly exceeds 95%. We can also pre-emptively warm the cache for known trending terms (the trending-topics service feeds the cache). For the small fraction that does miss, the index cluster has enough headroom to absorb 5-10× normal QPS without falling over.tweet_id → owning shard maintained as a side-effect of every index update. The recovering shard asks the Reverse Index "which tweets did I own?", bulk-fetches only those from Cassandra (a few billion, not 730), and re-tokenizes them through the Index Builder. Cuts recovery from days down to hours. Without the Reverse Index, a shard loss is a multi-day outage.