Search Service at Twitter (2014)

Ideas you will meet

Distributed Systems

Consensus, time, failure and the fundamentals of scale-out.

Search

Inverted indexes, relevance, autocomplete and search engines.

What we already know about this

Our own lessons on these ideas. Read them here before you start, so it makes sense.

• Search, Analytics and Geo · 11 minSharding a search index: scatter, gather and mergeHow a search index is split across machines, why queries fan out to every shard, how top-k results are merged, and what goes wrong with deep paging and uneven shards.Read hereClose

  Why shard

  One machine cannot hold or search a very large index within the latency budget. Sharding splits the index into pieces, each on its own machine, so storage and work are spread out. Replicas copy each shard for availability and extra read capacity.

  Before: one big index

  100%

  After: sharded, scatter and gather

  100%

  What changed

  • Storage and work are spread over machines
  • Each shard returns its own top results to be merged
  • The slowest shard decides how fast the query is

  Two ways to partition

  • Document-partitioned (local index): each shard indexes a subset of the documents and holds a complete inverted index for them. Writes touch one shard. Every query must visit every shard. This is what Elasticsearch and Solr do.
  • Term-partitioned (global index): each shard holds the postings for a subset of terms. A query touches only the shards for its terms, but a write touches many shards and long posting lists must be combined across machines. It is rare in practice.

  100%

  Figure SDA-SAG-05-1 · Scatter, gather and merge · © forcsguy.com

  Decisions that matter

  • Shard size and count. Too many tiny shards waste memory and add fan-out; too few huge shards recover slowly and cannot spread load. Many teams aim for tens of gigabytes per shard. The primary shard count is hard to change later, so plan for growth or use time-based indexes.
  • Routing. By default a document goes to a shard by hashing its id, which spreads data evenly. Routing by tenant or user keeps one customer's data together so queries can hit a single shard, at the risk of uneven shards.
  • Tail latency. A query is as slow as its slowest shard. Use replicas, timeouts, and hedged requests (send to a second replica if the first is slow).
  • Deep pagination. Page N needs the first N x pageSize results from every shard. Use cursors (search-after) instead of large offsets.
  • Scoring across shards. Each shard computes statistics such as IDF from its own documents. Even, large shards make the difference small; otherwise gather global statistics first.
  • Rebalancing and reindexing. Plan how to add capacity without downtime, usually by indexing into a new index and switching an alias.

  100%

  Figure SDA-SAG-05-2 · Time-based indexes · © forcsguy.com

  Keeping the index fresh

  Documents usually live in a primary database and are copied to the search index. Use a change stream or an outbox so updates are not lost, accept that search is eventually consistent, and keep a way to rebuild the whole index from the source of truth.

  Real-world reading

  Related reading

  • Elasticsearch: scalability and resilienceElasticDocumentation
  • SolrCloud: shards and indexing dataApache Solr · Apache Software FoundationDocumentation
  • Elasticsearch GuideElasticDocumentation
  • Scaling Elasticsearch Clusters at UberUber engineers · UberCase Studies

  Show 2 moreShow fewer

  • Elasticsearch Performance Tuning Practice at eBayeBay engineers · eBayCase Studies
  • Solr: Improving Performance for Batch Indexing at BoxBox engineers · BoxCase Studies

  Open this lesson with its quiz and exercise

• Search, Analytics and Geo · 10 minNearby search with Redis GEOHow Redis stores locations in a sorted set using geohash scores, how GEOSEARCH finds nearby members, how to expire stale entries, and when to move to a spatial database.Read hereClose

  The problem

  "Show the nearest available drivers", "stores within 3 km", "friends nearby". The data is a large set of points that change often, and the question is always about a small neighbourhood. You want a fast, in-memory structure.

  How Redis does it

  Redis stores locations in an ordinary sorted set. Each member (a driver id) has a score that is a 52-bit geohash of its longitude and latitude. Geohash interleaves the bits of the two coordinates so that nearby places have nearby scores. A search turns a radius into a few score ranges, reads those ranges, then filters by exact distance.

  100%

  Figure SDA-SAG-09-1 · A nearby query in Redis · © forcsguy.com

  Typical commands:

  GEOADD drivers:blr 77.5946 12.9716 driver:42
  GEOSEARCH drivers:blr FROMLONLAT 77.60 12.97 BYRADIUS 2 km ASC COUNT 20 WITHDIST
  

  Design details

  • Partition the key. One sorted set for the whole planet is a hot spot and slow to search. Use a key per city, region or grid cell so each set is small and local, and shard keys across nodes.
  • Updates are cheap. Moving a member is another GEOADD, about O(log n). 75,000 updates per second is manageable on a few nodes.
  • Freshness. Sorted set members have no individual expiry. Keep a second sorted set lastseen with the update time as score, and run a job that removes members older than, say, 30 seconds from both sets.
  • Availability filter. Store only drivers who are available, or look up status after the search; keep the hot set small.
  • Accuracy. Redis uses a spherical model of the Earth, so distances can be off by a fraction of a percent, and results near a cell boundary rely on the exact filter. That is fine for "within 2 km", not for surveying.
  • Durability. Treat Redis as a rebuildable copy. The system of record is a database, and clients re-send their location every few seconds anyway.

  100%

  Figure SDA-SAG-09-2 · Keeping the hot set fresh · © forcsguy.com

  When Redis GEO is not enough

  Move to a spatial database such as PostGIS, or a purpose-built index, when you need polygons and fences, joins with other tables, constraints (opening hours, vehicle type) together with distance, or durable queries over history. Many systems combine them: a database for rich queries and Redis for the live, hot position data.

  Other grid systems are covered in the geofencing lesson and in the real-world reading: H3 (hexagons, good for aggregation and neighbours) and S2 (cells on a sphere).

  Real-world reading

  Related reading

  • Redis geospatial data typeRedisDocumentation
  • Redis GEOSEARCH commandRedisDocumentation
  • PostGIS documentationPostGIS project · PostGISDocumentation
  • H3: Uber's Hexagonal Hierarchical Spatial IndexUber Engineering · UberCase Studies

  Show 1 moreShow fewer

  • S2 GeometryGoogle · S2 GeometryDocumentation

  Open this lesson with its quiz and exercise

• Classic System Designs · 10 minDesign search autocompleteSuggest completions as the user types, within a tenth of a second: tries, top-k per prefix and how suggestions are updated.Read hereClose

  The problem

  As a user types "syst", show a handful of likely completions: "system design", "systemd", "system of a down". The suggestions must appear almost instantly (under about 100 ms) or they are useless, and they should reflect what people actually search.

  Requirements

  • Fast: respond within ~100 ms.
  • Return the top 5 to 10 completions for the typed prefix, ranked by popularity.
  • Handle high volume (a request per keystroke).
  • Updated regularly with new trends.

  The core data structure: a trie

  A trie (prefix tree) stores words along paths of letters, so all words that share a prefix share a path.

  100%

  Figure SDA-CSD-09-1 · Trie with word frequencies · © forcsguy.com

  To find completions for a prefix, walk to its node, then look at everything beneath it. Doing that on every keystroke is too slow for a big trie, so we precompute: at each node store the top k most popular completions for that prefix.

  100%

  Figure SDA-CSD-09-2 · Top-k completions cached per prefix · © forcsguy.com

  Now a query is: walk the prefix (a few steps), read the stored list. That is fast and independent of how large the subtree is.

  From raw queries to counts

  100%

  Figure SDA-CSD-09-3 · Counting queries into a frequency table · © forcsguy.com

  Two services

  100%

  Figure SDA-CSD-09-4 · Online query path and offline trie build · © forcsguy.com

  Query service (online)

  1. Receive the prefix.
  2. Look it up in the in-memory trie or a prefix cache.
  3. Return the top suggestions.

  Data pipeline (offline)

  1. Search queries are logged.
  2. A periodic job (hourly or daily) aggregates counts per query, filtering out spam and abusive terms.
  3. A builder creates the new trie with top-k lists.
  4. The new version is published to the serving machines and swapped in.

  Real-time updates for every search would be costly and unstable. For breaking trends you can add a small fast path for recent spikes.

  Looking up a prefix in the trie

  100%

  Figure SDA-CSD-09-5 · Autocomplete lookup on a trie with cached top lists · © forcsguy.com

  When one query's count changes, every node on the path above it must refresh its stored list, which is why the trie is rebuilt in a batch rather than edited per search:

  100%

  Figure SDA-CSD-09-6 · Updating a count touches all ancestors · © forcsguy.com

  100%

  Figure SDA-CSD-09-7 · A trie stored as key-value pairs · © forcsguy.com

  100%

  Figure SDA-CSD-09-8 · Filtering unwanted suggestions · © forcsguy.com

  100%

  Figure SDA-CSD-09-9 · Sharding the trie by prefix · © forcsguy.com

  Making it faster and bigger

  • Client side: wait a short pause after typing (debounce), skip very short prefixes, and cache previous responses in the browser.
  • Server side: keep the trie in memory; cache hot prefixes; set a CDN or edge cache for the most common ones.
  • Limit prefix length (for example 50 characters) to bound work.
  • Shard the trie across machines, for example by the first letters, balancing by traffic not by alphabet (some letters are far more common).
  • Replicate the serving tier for availability.

  The numbers and the trie's cost

  Each keystroke is a request, so 10 million users doing 10 searches of about 20 keystrokes is on the order of 24,000 requests per second (about double at peak), while only roughly 0.4 GB of new query text arrives per day. The trie's walk to a prefix is proportional to the prefix length, then reading its children and sorting them would be slow; two changes make lookup effectively constant time: cap the prefix length, and store the top few suggestions on every node (trading space for time). The cost moves to updates: changing one count means refreshing the cached lists on every ancestor up to the root, which is why the trie is normally rebuilt in a batch (say weekly) from aggregated logs and swapped in, rather than edited per query.

  Storing the trie. Either serialise each weekly snapshot into a document store, or flatten it into a key-value store where each prefix is a key and the node's data is the value. Serve from a distributed cache; on a miss, fill the cache from the store.

  Log less. Sample the query log (record one in N) because logging every keystroke-level request is costly and top results are statistically stable.

  Take-down. Put a filter layer in front of the cache to hide unsafe suggestions immediately, and delete them from the data before the next rebuild.

  Variants. Different countries can have their own tries (even served from a CDN); other languages need Unicode in the nodes; trending queries need a faster path, such as stream processing and a ranking that weights recent searches more.

  Quality details

  • Ranking can combine popularity, recency and personalisation.
  • Safety: filter offensive or sensitive suggestions.
  • Typos: optionally tolerate small spelling errors.
  • Languages: tokenisation and ordering differ per language, so plan for it early if you need it.

  Worked numbers: does the whole thing fit in memory?

  Say the system serves the 10 million most common prefixes, storing the top 5 suggestions for each at roughly 100 bytes (five short strings plus scores). That is about 1 GB, comfortably inside one machine's memory, and a few GB with overhead. Even 100 million prefixes is only about 10 GB. So the "answer from memory, rebuilt in the background" design is realistic; no search over a huge dataset happens per keystroke.

  Latency budget for a 100 ms target as seen by the user: the browser debounce is an intentional pause (not part of the budget); network round trip 30 to 50 ms; edge or load balancer 5 ms; service 5 to 10 ms to hash the prefix and read the cached list; response and render 10 ms. There is little room for anything slower, which is why the lookup must be a single key read.

  Request volume: 10 million users x 10 searches x about 20 keystrokes is 2 billion prefix lookups a day, about 24,000 per second on average. Debouncing (waiting about 150 ms after the last keystroke) and not querying for very short prefixes can cut that by half or more.

  100%

  Figure SDA-CSD-09-10 · Debounce and ignore late answers · © forcsguy.com

  Ranking: popular, recent and personal

  Raw popularity is a good start but has two problems. It favours old favourites over trending queries, and it is the same for everyone. A common improvement blends signals:

  score = popularity x decay + trending boost + personal boost

  • Decay: weight recent searches more, for example by halving a query's weight every day or week, so a sudden event rises quickly and fades.
  • Trending: a fast path watches a stream of recent queries and lifts those whose rate jumps compared with their normal level.
  • Personal: a user's own recent searches, kept separately (and cached per user, not in the shared trie).
  • Safety and quality: filter offensive or illegal suggestions and anything with too few searches to be useful or private.

  100%

  Figure SDA-CSD-09-11 · Blending slow and fast ranking signals · © forcsguy.com

  Mistakes people make

  • Searching the data per keystroke. Precompute the top suggestions per prefix.
  • Ignoring out-of-order responses. A slow response for "sy" arriving after "syst" overwrites the list with stale results.
  • Putting personal data in the shared cache. Keep personal suggestions out of anything shared across users.
  • Forgetting normalisation. Case, accents, extra spaces and different Unicode forms must map to the same prefix.
  • No filtering. Autocomplete amplifies whatever people type, including offensive or private terms.
  • Rebuilding on every search. Rebuild in batches and swap versions atomically.

  Try it yourself

  Build the tiny search engine lab for the index side, and read the query understanding lesson for how typed text is cleaned before it is matched.

  Real-world reading

  Related reading

  • YouTube ArchitectureHigh ScalabilityCase Studies
  • Hello Interview: System Design in a HurryHello InterviewCourses
  • Hello Interview: System Design WalkthroughsEvan King & Stefan Mai · Hello InterviewYouTube
  • Meta Engineering BlogMeta Engineering · MetaBlogs

  Show 6 moreShow fewer

  • Cloudflare Learning Center: What is a CDN?CloudflareDocumentation
  • Solving Remote Build Cache Misses (2 parts) at SoundCloudSoundCloud engineers · SoundCloudCase Studies
  • Scaling Memcache at Facebook (paper)Nishtala et al. · Meta (Facebook)Papers
  • Reduce Memcached Memory Usage by 50% at TrivagoTrivago engineers · TrivagoCase Studies
  • Amazon Architecture - High Scalability -High Scalability engineers · High ScalabilityCase Studies
  • LINE Engineering BlogLINE engineers · LINEBlogs

  Open this lesson with its quiz and exercise