SplitNet Architecture for Ad Candidate Ranking at Twitter

Ideas you will meet

Distributed Systems

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

Search

Inverted indexes, relevance, autocomplete and search engines.

Machine Learning Systems

Recommendations, ranking, feature stores and serving models at scale.

What we already know about this

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

• Real-World Systems · 13 minHow a Twitter-style service worksPosting, timelines, search, trends and counters at huge scale: the architecture of a short-message social network.Read hereClose

  The problem

  A service where people post short messages, follow others, and read a timeline of recent posts. The numbers are what make it hard: hundreds of millions of users, huge read traffic, and a few accounts followed by tens of millions.

  Core operations

  • Post a message (with media, mentions, hashtags).
  • Follow and unfollow.
  • Read the home timeline (people I follow), and a user timeline (one person's posts).
  • Search posts and see trends.
  • Likes, replies, reposts and notifications.

  Estimates

  Say 300 million monthly users, 100 million daily active, each reading their timeline 10 times a day and posting a few times a week.

  • Timeline reads: 1 billion a day, about 12,000 a second on average and several times that at peaks.
  • Posts: about 50 million a day, a few hundred per second on average.
  • Reads outnumber writes by roughly 20 to 1, so we optimise for reading.

  High-level architecture

  100%

  Figure SDA-RWS-16-1 · Short-message service architecture · © forcsguy.com

  Posting

  1. The tweet service validates the post and obtains a unique, time-sortable id (Snowflake-style, see the ID generator lesson).
  2. It stores the tweet in a sharded tweet store, partitioned by tweet or user id. Media goes to object storage and is served through a CDN.
  3. It publishes a "tweet created" event to a stream, so the rest of the work happens asynchronously: fan-out, search indexing, notifications, trends and analytics.

  Home timeline: fan-out

  The home timeline is the hot path. Two strategies (from the news feed lesson):

  • Fan-out on write: when a user posts, workers look up their followers and push the tweet id into each follower's cached timeline list (for example Redis, keeping only the latest few hundred ids). Reading a timeline is then one quick lookup.
  • Fan-out on read: build the timeline on demand by merging recent tweets of everyone you follow. Cheap to post, expensive to read.

  The hybrid used at scale: fan out on write for most accounts, but skip accounts with huge follower counts. Their tweets are fetched at read time and merged into the cached list. This avoids one post creating tens of millions of writes.

  Reading the timeline

  1. Fetch the user's cached list of tweet ids.
  2. Merge in recent tweets from followed popular accounts, plus promoted content.
  3. Hydrate ids into tweets, authors and counts using caches (tweet cache, user cache, counters).
  4. Rank (recency, engagement, relationship, relevance) and return a page with a cursor.
  5. Filter muted, blocked and deleted items at read time.

  User timeline

  A user's own posts are a simple range query: tweets for one author in time order. They are stored by author id and cached, and paginated with a cursor.

  The social graph

  Who follows whom is a huge graph with fast lookups in both directions (followers of A, accounts followed by B). It is stored in a specialised, sharded store, with caches, and updated asynchronously when users follow or unfollow.

  Search

  Posts are indexed in an inverted index (term to list of tweet ids), usually built on a Lucene-style engine. The indexer consumes the event stream, so new tweets become searchable within seconds. The index is sharded by time (recent tweets are the most searched and kept in memory) and replicated. Queries combine text relevance, recency and engagement.

  Trends

  Trending topics are not the most common words, but the ones that are unusually frequent right now. A stream processor counts hashtags and phrases in sliding windows (say 5 minutes), compares them to their normal rate, filters spam and sensitive terms, and publishes the result per region.

  100%

  Figure SDA-RWS-16-2 · Search indexing and push notifications from the same event · © forcsguy.com

  Counters and notifications

  Likes, reposts and view counts are updated very frequently. Use sharded counters and approximate, eventually consistent values, batching updates, since an exact real-time number is rarely required. Notifications are produced from the event stream and delivered through the notification system (see that lesson).

  Scale and reliability

  • Cache everything hot: timelines, tweets, users, counters. The tweet store is the source of truth, not the read path.
  • Shard stores by id, and replicate across zones and regions.
  • Rate limit posting, following and API calls to fight spam and abuse.
  • Degrade gracefully: if ranking is slow, return a chronological timeline; if search lags, show fewer results.
  • Moderation and abuse systems run on the same event stream.
  • Real systems evolve, and details in public talks differ from year to year; the principles above (async fan-out, hybrid push and pull, caching, sharding) are the durable part.

  Real-world reading

  Related reading

  • Hello Interview: System Design in a HurryHello InterviewCourses
  • Hello Interview: System Design WalkthroughsEvan King & Stefan Mai · Hello InterviewYouTube
  • Consistent Hashing and Random Trees (paper)Karger et al. · MITPapers
  • The System Design PrimerDonne Martin · donnemartinGitHub

  Show 6 moreShow fewer

  • How Discord Stores Trillions of MessagesBo Ingram · DiscordBlogs
  • How Discord Stores Billions of MessagesDiscordCase Studies
  • YouTube ArchitectureHigh ScalabilityCase Studies
  • TAO: The power of the graphMeta Engineering · MetaCase Studies
  • Dropbox Tech BlogDropbox engineers · DropboxBlogs
  • Productionizing ML with Workflows at TwitterTwitter engineers · TwitterCase Studies

  Open this lesson with its quiz and exercise

• 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