Sharded Counters

It is 10:40 pm and India have just won the cricket final. Minutes later the winning captain posts a photo from the dressing room. In ten minutes it collects 2 million likes, about 3,300 every second, and 5,000 a second in the busiest minute. Each like adds one to the post's like count, which shows under the photo for millions of viewers. On a normal evening one database row holds that number easily. Tonight the like button spins, then fails, and the count freezes at 41,000 while fans keep tapping. Nothing is wrong with the database as a whole: one row is drowning.

The captain's photo is post 42. The obvious design keeps its count in one place and runs UPDATE posts SET likes = likes + 1 WHERE id = 42 for every tap. On a normal day that is fine. For this post it is not, because every one of those updates is aimed at the same row.

A hot key is a single key or row that gets far more traffic than the rest. Partitioning the database does not help here: post 42 lives on one partition, so one machine takes the whole burst while the others sit idle.

Why does one row give up at a few thousand likes a second? Think of a single ticket window at a railway station: however fast the clerk works, everyone waits in one line. To add one safely, the database locks the row, reads the value, writes the new value and commits, and the commit often waits for the log to reach disk or a replica. Every other update to that row waits for the lock, so the time one update holds it sets a hard ceiling:

Durability is what costs: the more each commit waits for, the lower the ceiling. Post 42's 3,300 likes a second, 5,000 in the busiest minute, swamp every row that waits for a disk and reach the ceiling of the one that does not.

Once likes arrive faster than the ceiling, the line of waiting updates grows every second. Latency climbs from milliseconds to seconds, requests time out, and clients retry, which adds even more load. Other writes to the same row, such as a caption edit, wait in the same line.

An in-memory store with an atomic increment command handles far more on one key, often 100,000 or more increments a second, because nothing waits for a disk. That is a fair answer for many apps. Sharding is still needed when the count must be durable on disk with every like, when one key's burst outgrows even that machine, or when the store has no atomic increment. Sharded counters work in any store.

If one row is the line, open more windows. The fix is to stop having one counter. Replace post 42's counter with N shard rows, each holding part of the count. A write picks one shard at random and adds one to it. A read adds up all N shards.

import random
N = 4  # shards for this counter

def add_like(post_id):
    shard = random.randrange(N)  # 0, 1, 2 or 3
    db.execute("UPDATE like_shards SET count = count + 1"
               " WHERE post_id = %s AND shard = %s", (post_id, shard))

def like_count(post_id):
    return db.query_one("SELECT SUM(count) FROM like_shards"
                        " WHERE post_id = %s", (post_id,))

Create a counter's shard rows when the counter is created, so every UPDATE finds its row.

With 4 shards, each row receives about a quarter of the updates, so the same database absorbs roughly four times the write rate on post 42. The total stays exact: every +1 lands in exactly one shard, so the sum of the shards is the true count. Why pick at random? It needs no coordination, and at thousands of writes a second each shard's share stays within a few percent of a quarter. Hashing the user id works as well, and keeps a user's like and unlike on the same shard.

Here is post 42 twice: first with one counter row, then with four shard rows.

The burst did not get smaller; it got spread. Each shard takes a quarter of the writes, and a read pays for it by touching four rows instead of one. Since a popular post may be viewed 50 times for every like, the cached total in the last step is what keeps that trade affordable.

Writes are handled; now the number under the captain's photo. There are three common ways to produce it.

Many apps combine them. The post page reads a cached total that is rebuilt from the shards when it expires. Feeds read the rolled-up number stored with each post, so a feed of 20 posts costs 20 reads, not 20 × N.

How many shards did post 42 need? Choose N from the peak write rate you must absorb and the rate one shard row takes comfortably, leaving headroom below its ceiling:

Each shard adds about 500 writes a second and one row to every uncached read: 4 shards take about 2,000 a second, 16 about 8,000, 64 about 32,000, and an uncached read of 64 shards touches 64 rows. Sixteen shards cover post 42's 5,000-a-second minute with room to spare.

Do not give every counter 64 shards. Most posts get a handful of likes a day, and multiplying their rows and reads by 64 is pure waste. Traffic is heavily skewed: a tiny fraction of posts gets nearly all the bursts, which is why N should grow per counter, only where needed.

Start every counter with one shard and add shards only to counters that get hot. Store N with the counter, for example in a small config row, and let writers cache it for a few seconds.

Raising N is safe because a new shard starts at zero and readers sum whatever rows exist. Lowering N takes care: publish the smaller N first, wait until no writer still uses the old N, then move each extra shard's count into shard 0 and delete that shard in one transaction. It is often simpler never to shrink, because an old post with 16 quiet rows costs very little. Where the shard rows live matters too:

Start with shards on one partition; spread them only when the machine itself, not just the row, is the bottleneck.

Sharded counters are not the only way to absorb post 42's burst. Each alternative gives up something different:

They combine well. A common pattern writes each like to an event stream so nothing is lost, and lets consumers add up a second of likes before writing one increment. Batching cuts the writes so much that one row, or a few shards, can take even a viral post. The message queue chapter covers the stream part.

Fans abroad like the captain's photo too. An app that serves users from two regions, say Mumbai and Singapore, takes likes in both. Sending every like to one home region adds a cross-region round trip to each tap. Keeping one total per region and copying it across with last write wins, from the key-value store chapter, loses likes: both regions start at 1,000, Mumbai adds 5 and Singapore adds 3, and whichever copy lands last, 1,005 or 1,003, wins. The true count is 1,008.

Treat each region as a shard instead. Each region adds only to its own entry, copies its entries to the other regions in the background, and shows the sum:

When two copies of an entry meet, keep the larger one: an entry only ever grows, so the larger value is the newer one. No cross-region locks, no lost likes, and each region's total is usually a few seconds behind. This is a grow-only counter, one of the CRDTs the key-value store chapter mentioned. Unlikes go in a second grow-only counter of removals, and the count shown is likes minus removals. Inside each region, a hot entry can still be split into shard rows.

Is post 42's count even supposed to be exact? Three different promises hide behind the word "count". Mixing them up leads to designs that are either over-built or quietly wrong.

A sharded counter with a cached total is exact, eventually: nothing is lost, and the displayed number trails reality by the cache time. Nobody can tell 2,104,331 likes from 2,104,977, least of all when the app shows "2.1M".

Counting distinct things is a different problem: knowing how many unique users viewed a post means remembering who was already counted. A HyperLogLog sketch estimates distinct counts without storing the ids. A common setup uses 16,384 small registers, about 12 KB, and its typical error (the standard error) is about 0.81%, so most estimates land within 2% of the truth. The size stays the same whether the true answer is a thousand or a billion.

Counting often comes with a ranking: the top 10 players in the live quiz during the final, the most-liked posts this hour, the hashtags trending as the match ends. These are top-K questions, and the right tool depends on whether ranks must be exact and how many distinct things you are counting.

Exact leaderboards: a sorted set. An in-memory store's sorted set keeps members ordered by score. ZINCRBY quiz:final 10 user:9001 adds 10 points, ZREVRANGE quiz:final 0 9 returns the top 10, and ZREVRANK gives one player's rank, each in time that grows with the logarithm of the set's size. A quiz with 2 million players fits easily on one node. If updates to one board outrun that node, batch them for a second, or split players across several sets and merge each set's top 10: the overall top 10 must be among them.

Heavy hitters in a huge stream: a count-min sketch plus a heap. Exact counts for millions of hashtags need memory for every one of them. A count-min sketch is a small grid of counters, say 5 rows of 10,000. Each hashtag is hashed once per row and adds 1 to one counter in each row; its estimate is the smallest of its 5 counters. Other tags sharing a counter only ever add to it, so an estimate can be too high, never too low. A min-heap of the current top 100 sits beside it: when a tag's estimate beats the smallest in the heap, it takes that place.

That grid is 200 KB however many tags there are. With 10,000 counters per row, an estimate is at most about 0.03% of all counted events too high, with about 99% confidence: under 30,000 for 100 M events. That is invisible next to a trending tag's millions and useless for ranking tags used 50 times, which is fine, because top-K needs only the big ones.

Trending: counts per time window. "Most liked this hour" needs counts that forget. Keep one bucket per minute, sum the last 60, and drop the oldest as each new minute starts. Count-min sketches of the same shape add cell by cell, so 60 minute-sketches make an hour-sketch, and sketches from 20 servers merge the same way. Trending usually means far above the usual count for that hour, so compare each window with the same hour on earlier days.

A like is not just +1. The same user must not count twice, whether they double-tap or their app retries after a timeout. So the real write has two steps: record that this user liked this post, then add one.

-- 1. Record the like; the primary key (post_id, user_id) rejects a repeat.
INSERT INTO likes (post_id, user_id, liked_at)
VALUES (42, 9001, now())
ON CONFLICT DO NOTHING;

-- 2. Only if step 1 inserted a row: add one to a random shard.
UPDATE like_shards SET count = count + 1
 WHERE post_id = 42 AND shard = 3;

PostgreSQL syntax. Run both steps in one transaction where the store allows it; an unlike deletes the row and subtracts one.

If the two steps cannot share a transaction, the count can drift: the like is stored but the increment is lost, or the reverse. A reconciliation job recounts the likes table for recently active posts and corrects the shards. The likes table is the truth; the counter is a fast summary of it.

This also answers "did I like this post?", which the app needs to colour the heart. That lookup reads one row of the likes table by (post_id, user_id) and never touches the counter.

The likes table has a hot spot of its own. Partition it by post_id and every one of the viral post's 3,300 inserts a second lands on one partition, even though no row is locked twice. Partition by a hash of (post_id, user_id) instead: the inserts spread over all partitions, and "did I like this?" is still one read, because both ids are known. The query that gets harder is "who liked this post", which must now visit every partition, so keep it rare and paginated, or build it from the like events in the background.

Sharded counters are simple, so most of their failures come from choosing N badly or forgetting the read path. Here is what each looks like during a burst like post 42's:

Most rows share one fix: the likes table and the cached total are the safety net, so a counter can be rebuilt, grown or shown stale without ever showing a wrong number for long.

Sharded counters usually come up inside a bigger design, such as likes, views or votes, and these lines carry the idea in under a minute.