Sharded Counters for the Top-K Problem

In one line: this is where a counting primitive becomes a product feature. Trends are just counters plus a threshold plus a time window — and the regional decomposition is what makes it tractable.

Counters per hashtag

For each hashtag, the system creates a counter and selects the shard count based on projected write volume — for example using follower counts as a proxy for expected traffic. Subsequent hashtag events update the existing sharded counter.

Hashtags are heavy hitters with an even worse distribution than tweets

Lesson 1's problem, intensified. A tweet's likes come from people who saw that tweet. A hashtag's uses come from everyone posting about that topic, across all of Twitter, simultaneously.

So the skew is more extreme: during a major event, one hashtag can absorb an enormous share of global write volume while millions of others sit idle.

It also means the prediction problem from Lesson 3 is harder. A tweet's expected engagement correlates with its author's follower count. A hashtag has no author — anyone can use it, and a hashtag nobody has seen before can explode in minutes.

The source's answer is to use follower count of the associated tweets as a proxy, and to note that "hashtags associated with popular or celebrity tweets may also receive sharded counters." That is reasonable and clearly weaker than the tweet case — which makes Lesson 4's dynamic resizing the primary mechanism here rather than the backstop.

Regional and global counters

Trends are primarily computed based on regional popularity. To support this, Twitter maintains multiple counters per hashtag, in addition to a global counter.

Global hashtag counter
        120M
          |
   +------+------+------+------+
   |      |      |      |      |
  760K   156K   130K    5K     3K      <- location-based counters
   |      |      |      |      |
 [shards][shards][shards][shards][shards]

• The global hashtag counter aggregates all location-based counters.
• Location-based counters track regional usage and compare it against a predefined threshold within a given time window to identify trending hashtags. If the threshold is 10,000, a hashtag becomes visible in the trends timeline for users in that region once the count is reached. If similar thresholds are reached across multiple regions, the hashtag may appear globally.

Regional counters are not just a feature — they are what makes the write path scale

It is easy to read the regional breakdown as purely a product requirement: people want local trends. It is also a structural necessity.

A single global counter for a trending hashtag would concentrate the world's writes onto one counter's shards. Splitting by region means writes from New York touch New York's counter and writes from Tokyo touch Tokyo's — geographically local writes hitting geographically local shards.

That gives three things at once:

• Write locality. No cross-region round trip on the write path.
• A natural sharding dimension that already correlates with where traffic originates.
• The product feature — regional trends — for free.

So the hierarchy is a case where the product requirement and the scaling requirement point the same way. That is worth naming in an interview, because it is unusual and it makes the design feel inevitable rather than assembled.

Note the global counter is derived — "aggregates all location-based counters" — so nothing writes to it directly. It is a rollup, not a hot key.

A threshold plus a time window is what turns a count into a trend

A raw count does not identify a trend. #weather might have a permanently high count and never be interesting; a hashtag going from 0 to 10,000 in an hour is.

Two parameters convert one into the other:

The time window makes it a rate rather than a total. Counting within a window is what distinguishes "popular" from "suddenly popular."

The threshold makes it a decision. Above 10,000 in the window, show it; below, do not.

Both are tunable per region, which matters — 10,000 uses is enormous in a small market and unremarkable in a large one. Fixed global thresholds would make small regions never trend and large ones always trend.

Note this is a read-then-decide pattern, which Lesson 6 warned about. It is safe here only because the decision is advisory — showing a hashtag slightly early or late is harmless. If crossing the threshold triggered a payment, the relaxed consistency would be disqualifying.

Merging local into global

From Lesson 8's storage discussion:

The same storage can support region-wise Top K trends: local Top K lists are sent to an application server, which merges them to produce a global Top K list.

Merging local Top-K lists does not reliably give the true global Top-K

This is a classic and genuinely subtle problem, and it is worth being able to state.

Suppose each region reports its top 10. A hashtag ranking 11th in every region appears in no local list — yet summed across all regions it might outrank hashtags that topped a single one.

So merging local Top-Ks produces an approximation. A hashtag that is broadly moderate everywhere can be invisible, while one that is enormous in one region reliably surfaces.

Standard mitigations, in increasing cost:

• Report top K′ where K′ > K — each region sends its top 100 to compute a global top 10. Widens the net; still not exact.
• Report counts for any hashtag in any region's list, so a hashtag surfacing anywhere gets its global total computed properly.
• Compute globally, which forfeits the write locality the whole design was built for.

For trends the approximation is fine — nobody can verify that the true 9th-place global hashtag was omitted. But knowing that the merge is lossy is exactly the kind of detail that separates a considered answer.

This is the same fan-out-and-merge shape as distributed search's merger, and it has the same caveat: local top-K plus merge is an approximation of global top-K, not a computation of it.

Top-K tweets

Consider the Top K tweets in a user's home timeline. These include tweets from followed accounts, liked tweets, and retweets from followed users. Tweets are ranked based on factors such as follower count and recency. Twitter may also surface promoted and popular tweets from non-followed accounts, based on location and overall popularity.

Note: Twitter also uses other metrics to optimize Top K selection, but we've discussed the leading metrics here.

Counters feed ranking; they are not ranking

Worth keeping the boundary clear. Sharded counters answer "how many?" — likes, retweets, uses. Ranking a timeline needs much more: recency, the viewer's relationships, predicted engagement, promotion, diversity.

So a counter is an input to ranking, not the ranking itself. The Distributed Search chapter made the identical point: retrieval is the easy half, ranking is where the system actually lives, and it needs signals the index does not hold.

The honest scope for this chapter: sharded counters make the count available at scale; what you do with the count is a different problem.

The regional split is doing two jobs at once, which is worth naming: it keeps writes local so no increment crosses a region, and it produces per-region trend data as a by-product rather than as a second pipeline.

Interview signal by level

Next: where the shards actually live.