Heavy Hitters and Top-k Trends

In one line: the same power-law distribution that broke fan-out in Lesson 5 breaks counters here, and the fix is structurally identical — spread the load and pay a merge later.

The heavy-hitter problem

Public figures with millions of followers generate massive traffic spikes when they tweet. This is known as the heavy-hitter problem. A single counter cannot handle the write throughput required for millions of simultaneous likes or views.

A counter is a single row, and a single row is a single lock

The failure is specific and worth stating precisely.

Incrementing a counter means a read-modify-write on one record. To keep it correct under concurrency, the store must serialize those operations — a lock, a compare-and-swap, or a single-writer partition.

Which means the counter's throughput ceiling is one machine's ability to serialize updates to one row. Perhaps thousands per second.

Now a celebrity tweets and a million people like it within a minute:

1,000,000 likes / 60 seconds = ~16,700 increments per second
                               ON ONE ROW

Every one of those requests contends for the same lock. The successful ones are slow; the rest fail or queue. And the contention does not just slow the counter — it can consume connections and threads across the whole tier.

Sharding by key does not help, which is the crucial point. Consistent hashing, range partitioning, and every other partitioning scheme in this course distribute different keys across nodes. Here there is one key, and it is hot.

When the hot spot is a single key, key-based partitioning has nothing to work with. You have to split the key itself.

Sharded counters

To address this, Twitter uses sharded counters. The system splits the counter for a specific tweet into multiple shards across distributed nodes. This allows the system to handle burst write requests in parallel.

One key becomes N keys, and the merge moves to read time

The mechanism is simple and the consequence is not.

BEFORE:  tweet_123_likes                    -> 16,700 writes/sec on ONE row
AFTER:   tweet_123_likes_shard_0 .. _N      -> 16,700/N writes/sec each
READ:    sum(all shards)

With 100 shards, each takes 167 writes per second — comfortable. A like picks a shard at random, or by the liker's ID, or — per Lesson 4's user_location parameter — by geography.

What this trades is the read. Where a count was one lookup, it is now N lookups plus a sum. That is the right direction, because Lesson 3 established the system is read-heavy — so shouldn't the expensive side be the write?

It is affordable because the read cost is paid once and cached. The aggregated count is computed periodically and served from cache; it does not re-sum on every view. So the merge happens far less often than either the writes or the reads.

This is the same structural move as Lesson 5's fan-out hybrid: take a hot single point, spread it across many, and pay a merge at a moment you choose. Both are responses to the same power-law distribution.

Sharded counters make the count eventually consistent, and the design is honest about it

Sharded counters improve performance by placing counter shards close to users, similar to a CDN. This reduces latency for interactions like views or likes. However, to balance load, the total count displayed on a tweet may update with a slight delay (eventual consistency) as the application server aggregates data from various regional shards.

Two things here, and the second is the interesting one.

Geographic sharding — a like from Tokyo increments a Tokyo shard. That is why Lesson 4's likeTweet carries user_location, and it means the write never crosses an ocean. The CDN analogy is apt: put the write where the user is, reconcile globally later.

The count is therefore eventually consistent, and the design says so plainly. The number under a tweet is a recent aggregate, not a live total, and it can differ between regions for a while.

That is exactly right for this data, and the reasoning is Lesson 2's: the cost of staleness is what decides the guarantee. Nobody can tell whether a viral tweet has 1,204,338 or 1,204,502 likes. The number is decoration.

But notice the tension with Lesson 2's read-your-own-writes requirement: "a user must see their own like instantly." Those are compatible only if you separate two things:

"Did MY like register?"   -> must be immediate  -> local state, one shard
"What is the TOTAL?"      -> may be stale       -> aggregated across shards

The client shows your like as applied immediately and the total catches up. Session-scoped consistency for your own action, eventual consistency for the aggregate — which is precisely the split Lesson 2 specified, implemented here.

Top-k trends

Sharded counters also support the Top-k problem, used to calculate trends. Twitter displays Top-k trends (hashtags) based on frequency:

• Local trends: hashtags popular within the user's geographic location.
• Global trends: hashtags popular worldwide.

The system uses a sliding window to identify hashtags with the highest frequency over time.

Time ->

#life      ####################
#food      ###############
#summer    ############
#science   #########
#art       ######
           ^ as the window slides, #life drops out
             and #music, #gaming, #travel enter the top-k

The sliding window is what makes trends 'trending' rather than 'popular'

The word doing the work is sliding.

A cumulative count would rank hashtags by all-time popularity, and the answer would never change — the same generic tags would win forever. Nobody wants that.

A sliding window counts only the recent past, so a hashtag that spikes in the last hour outranks one that has been steadily used for years. The design's illustration shows exactly this: as the window advances, #life falls out of the top-k and #music, #gaming, and #travel enter.

This is genuinely harder than it looks, for two reasons.

You cannot just add — you must also subtract. As the window slides forward, counts from the trailing edge must be removed. Naively that means storing every event with its timestamp, which at Lesson 3's volume is enormous.

The standard resolution is bucketing: divide the window into fixed intervals, keep a count per bucket per hashtag, and slide by dropping the oldest bucket and adding a new one. The window becomes approximate at the granularity of a bucket, which nobody notices.

Exact top-k over a huge key space is expensive. Tracking the count of every hashtag to find the top ten means holding millions of counters. Production systems use sketch structures — count-min sketch and similar — which give approximate counts in bounded memory, with error concentrated on the rare items you were never going to return anyway.

Approximation is safe when the errors land on items outside the answer. That is the same admissible-error reasoning as that building block's ANN search and that building block's aerial-distance pruning: design the approximation so it fails in the direction that costs nothing.

Local versus global trends is the same aggregation tree as the counters

Local and global trends are not two systems — they are two levels of the same aggregation.

per-region counts  ->  local trends  (read directly)
                   ->  summed        ->  global trends

Which reuses exactly the machinery the sharded counters already need: regional shards that aggregate upward. Local trends read a region's shard; global trends read the sum.

Two consequences worth naming:

Local trends are fresher than global ones, because they skip the aggregation step. That is the right way round — local trends are the more time-sensitive product.

Global trends can be dominated by the largest region unless normalized. A hashtag trending across a huge user base will outrank a genuine spike in a smaller country. Real systems weight by rate of change rather than raw volume, which is why a hashtag "trends" when it accelerates, not when it is merely common.

A trend is a derivative, not a level. That single reframing explains most of what is otherwise puzzling about trending algorithms.

How this connects back to the timeline

The home timeline displays a stream of tweets from followed accounts, along with retweets and likes. The system aggregates interaction counts from sharded counters to determine which content to display.

Worth noting because it puts the counters on the read path of Lesson 5's timeline assembly.

Recall from Lesson 5 that timelines store IDs, not tweet bodies, precisely so that counts stay fresh. This is the payoff: at read time the system hydrates each tweet ID with its body from cache and its engagement counts from the counter service.

And the counts are not merely decorative — the design says they "determine which content to display," meaning engagement feeds ranking. So the counter service is not a display detail; it is an input to what appears in the feed at all.

That raises the stakes on its latency. A component that only decorated the output could be slow or fail open. One that determines the output cannot.

Two window shapes, and why one is much easier

"Top trends in the last hour" hides a choice most designs skip past.

Tumbling windows are cheap because every event belongs to exactly one bucket, so aggregation is a plain increment. The cost is that the answer changes in steps — at 10:00 the last hour resets, and a topic trending at 9:59 vanishes.

Sliding windows are what users actually expect, and the implementation is the useful trick: keep fine-grained buckets — per minute — and maintain the hour total by adding the newest minute and subtracting the minute that just fell out of range. You never re-scan the window; you adjust it at the edges.

That reduces a continuously-moving aggregate to two arithmetic operations per tick, which is what makes sliding windows affordable at this event volume.

Approximate counts, corrected by a slower exact pass

Probabilistic counting buys enormous memory savings and gives up exactness. That is usually the right trade for trends — nobody cares whether the ninth trending topic had 40,100 or 40,300 mentions — but it is worth pairing with the standard remedy.

Run a slower exact pass over the same window and overwrite the approximation when it completes. The fast path stays fresh and cheap; the slow path repairs drift. Users see an approximate answer immediately and a correct one shortly after, and no single layer has to be both fast and exact.

Next: why Twitter abandoned centralized load balancing.