Sharded Counters: Counting Likes at Scale

Every like for that post updates the same single number. That number lives on one machine, and one machine can take only so many writes a second. Past that limit, likes fail while the rest of the database sits idle. Sharded counters fix this by splitting that one number into many small ones.

Picture a common setup. A short-video app from Mumbai shows a heart and a like count under each video, called a reel. The likes service keeps one row per reel holding its count, and each tap adds 1 to that row. The setup is made up, but the problem is real.

On a normal day, a popular reel gets maybe 50 likes a second, and nothing complains. Then a famous playback singer posts a 20-second reel of herself singing on Marine Drive at sunset.

Within ten minutes, it is getting 9,000 likes a second. The service starts returning errors. The heart turns red on the phone, then quietly turns grey again.

The strange part is the dashboard. The database as a whole is mostly idle. One machine, the one holding that single reel's row, is at its limit, and every other machine is waiting for work.

You need five words: key, partition, hot key, atomic increment and shard. Each one gets a plain meaning and an example from the app.

• Key: the name you use to find one piece of data in a store. The singer's like count lives under the key reel#8841#likes.
• Partition: one slice of a big database that lives on one machine or a small group of machines. Many databases pick a key's partition by hashing the key, turning it into a number.
• Hot key: one key that gets far more traffic than the rest. The singer's reel is a hot key, and its partition is the hot partition.
• Atomic increment: adding a number to a value as one step that cannot be cut in half. Two likes arriving together both count, and neither overwrites the other.
• Shard: one piece of something that was split up. Here, a shard is one small counter that holds part of a reel's likes.

Adding machines spreads different keys around, but one key still lives in one place. All the writes for that key still go to one partition, however big the cluster gets.

Real databases document this limit. Amazon DynamoDB, a managed key-value database, says each partition is designed to deliver at most 1,000 write units a second (as of October 2026). One write unit is one write a second of an item up to 1 KB.

DynamoDB does try to help. Its adaptive capacity can move a very busy item onto a partition of its own. Even then, its docs say that item gets at most the partition maximum of 1,000 write units a second.

It also keeps up to five minutes of unused capacity as burst capacity, which absorbs short spikes. A reel that stays viral for hours outruns both.

The app's store has a similar per-partition cap. Here is what happened on the viral evening. Look at the one busy partition and the many quiet ones.

Think of a highway toll plaza with only one booth open. A thousand trucks arrive at once, and the queue backs up for kilometres. Widening the highway does nothing, because every truck still has to pass that one booth.

The booth is the one counter key, each truck is a like, and widening the highway is adding machines. Where the picture breaks: a real queue waits, but a database rejects writes over the cap, and the app must retry or drop them.

Sharded counters store one count as many small counters, each under its own key. A write adds 1 to just one of them, and a read adds all of them up.

Go back to the toll plaza. Now it opens 20 booths. Each booth keeps its own tally of trucks on a clicker. At the end of the shift, a supervisor walks past every booth and adds the 20 tallies.

Sharded counters work the same way. The likes service gives the singer's reel 20 counters, with keys reel#8841#likes#0 up to reel#8841#likes#19. In a store that hashes keys, each key is hashed on its own, so the 20 keys spread across partitions.

They spread only as far as the store has partitions, so several shards may share one at first. DynamoDB's docs say adaptive capacity then moves busy items apart, but not how fast. So watch for rejected writes right after adding shards.

This is a documented pattern, not a trick. The DynamoDB docs call it write sharding: add a random number to the end of a busy key so writes spread out. They warn that reading everything back means querying every suffix and merging the results.

Google's Firestore, a managed document database, describes the same idea in its docs as distributed counters. Each counter is split into shards, an increment picks a random shard, and the count is the sum of the shards.

Here is the write path after the change. Each like picks one shard at random. The likes service also saves a small record of who liked what, in a separate table.

On average, 9,000 likes a second over 20 shards gives each shard 450 writes a second. Random choice is a little uneven, so the busiest shard gets more than that, as the check below shows.

You pick one shard at random and run an atomic increment on that shard's key. Before that, you record the like itself, so the same person is not counted twice.

In the code below, db is a stand-in for the database client. Its save_like_once helper writes a row for this user and reel only if none exists yet, and returns False if one already did. Its increment helper adds a number to one key atomically.

import random

NUM_SHARDS = 20

def shard_key(reel_id, shard):
    return f"reel#{reel_id}#likes#{shard}"

def add_like(db, user_id, reel_id):
    if not db.save_like_once(user_id, reel_id):   # already liked: count nothing
        return
    shard = random.randrange(NUM_SHARDS)          # pick a booth from 0 to 19
    db.increment(shard_key(reel_id, shard), 1)    # +1 on one small key

Two lines matter most. random.randrange(NUM_SHARDS) picks a number from 0 to 19, so writes spread across all 20 shards. Over thousands of likes, each shard ends up with roughly the same share.

The first if line matters just as much. A fan who double-taps, or whose phone sends the like twice, is caught here. The who-liked-what table is keyed by user, not by reel, so it spreads across partitions on its own.

You fetch every shard and add the values together. Because that costs one read per shard, you cache the total for a second or two instead of summing on every view.

Here is the answer to the prediction. 50,000 views a second times 20 shards is 1,000,000 shard reads every second, just to show one number.

The toll plaza solves this the same way. The supervisor does not walk to all 20 booths each time a manager asks for the total. He walks the line every few minutes and writes the latest total on a board, and anyone who asks reads the board.

Follow the read path below. Start at the viewer. The No branch, which sums the shards, should run rarely, even on the viral reel.

The code uses shard_key and NUM_SHARDS from the first block. get_many is a stand-in that fetches many keys in one batch request. DynamoDB's BatchGetItem, for example, returns up to 100 items in one call.

import time

CACHE_SECONDS = 2
cache = {}  # reel_id -> (total, when it was summed)

def like_count(db, reel_id):
    saved = cache.get(reel_id)
    now = time.monotonic()
    if saved and now - saved[1] < CACHE_SECONDS:
        return saved[0]                            # the board, no shard reads
    keys = [shard_key(reel_id, s) for s in range(NUM_SHARDS)]
    values = db.get_many(keys)                     # a shard never written is None
    total = sum(v or 0 for v in values)
    cache[reel_id] = (total, now)
    return total

The if line is the cache check: a total younger than 2 seconds goes straight back. The v or 0 part treats a shard that no like has touched yet as zero, instead of crashing.

Now redo the maths. The app runs 40 likes servers. Each sums the viral reel at most once every 2 seconds, so that is 20 sums a second, or 400 shard reads instead of 1,000,000.

One small touch keeps the fan happy. When you tap the heart, the app adds 1 to the number on your own screen right away. You see your like land, even though the shared total catches up a moment later.

Check it yourself

Add this third block after the first two, save all three as one file, and run it with Python 3. MemoryDB is a stand-in store that lives in memory, so nothing else is needed.

class MemoryDB:                       # a stand-in store that lives in memory
    def __init__(self):
        self.likes, self.counts = set(), {}

    def save_like_once(self, user_id, reel_id):
        if (user_id, reel_id) in self.likes:
            return False
        self.likes.add((user_id, reel_id))
        return True

    def increment(self, key, n):
        self.counts[key] = self.counts.get(key, 0) + n

    def get_many(self, keys):
        return [self.counts.get(k) for k in keys]

random.seed(7)
db = MemoryDB()
for user in range(9000):              # one second of the viral reel
    add_like(db, user, 8841)
add_like(db, 42, 8841)                # a double tap from user 42
shards = db.get_many([shard_key(8841, s) for s in range(NUM_SHARDS)])
print("total likes:", like_count(db, 8841))
print("busiest shard:", max(shards), "quietest shard:", min(shards))

On Python 3.13 it prints total likes: 9000, then busiest shard: 481 quietest shard: 393. The double tap from user 42 added nothing, and no shard got close to the 1,000-a-second cap.

What just happened? Every shard got some likes, but not the same number. The busiest got 481 against an average of 450. That gap is why the next section plans for less than the full cap per key.

Divide your planned peak writes a second by a safe write rate for one key, well below its cap. More shards make writes safer but make every uncached read cost more.

1. 1

   Find the peak write rate

   The biggest viral reel so far peaked at 9,000 likes a second. The team plans for 14,000, about one and a half times that, because the next one could be bigger.

2. 2

   Pick a safe rate per key

   The store caps a partition at about 1,000 writes a second. The team plans for 700, leaving 30 percent for random unevenness between shards.

3. 3

   Divide and round up

   14,000 divided by 700 gives exactly 20 shards. If the division leaves a remainder, round up.

4. 4

   Check the busiest shard

   In a simulation of 2,000 seconds at 14,000 likes a second, the busiest of 20 shards peaked at 810 writes (made-up load). That stays under the 1,000 cap.

5. 5

   Check the read cost

   Twenty small items fit in one batch read. Summing them every 2 seconds per server is cheap, so 20 shards is fine on the read side too.

Why not plan for the full 1,000 per key? The same simulation at 18,000 likes a second over 20 shards gave a busiest shard of up to 1,019 writes. An average of 900 looks safe, but the unlucky shard crosses the cap.

Headroom does not cover shards that share a partition. Two shards at 700 each send 1,400 writes a second to one partition. That is why the team watches for rejected writes after every grow.

Not every reel needs 20 shards. Most reels get a few likes a minute, and giving each one 20 shards means 20 items to store and read for no benefit. So the team stores the shard count per reel, in a small settings item next to the counter.

Every reel starts with 1 shard. A small monitor watches the like rate per reel, and when a reel crosses 300 likes a second, it raises that reel to 20 shards. Quiet reels keep one, like quiet roads that keep one booth open.

An increment is not idempotent. Idempotent means doing it twice has the same effect as doing it once. If an increment times out, the service cannot tell whether it was applied, so any retry is a guess.

Database docs say this plainly. DynamoDB's docs say atomic counter updates are not idempotent, so a retry can overcount. They suggest such counters for things like website visitors, but not for a banking application.

The docs for Apache Cassandra, an open-source distributed database, say the same about its counter columns. After a timeout, the client cannot know if the update was applied, and replaying it may count it twice.

The add_like code above has a quieter failure too. It saves the like first, then increments. If the increment fails and the caller retries add_like, save_like_once returns False, and the like is never counted.

For likes, the team accepts a small drift on the live count, because a like count is not money. The who-liked-what table is the truth. A nightly job counts rows per reel and adds the difference, which can be negative, to one shard.

It adds the difference as an increment instead of overwriting, so likes landing during the job are not lost. A like that arrives mid-run may leave a tiny new gap, which the next run closes.

Making each like count exactly once

If drift is not acceptable, write the like row and the shard increment in one transaction, a group of writes that all happen or none do. In DynamoDB, TransactWriteItems can do both, with a condition that the like row does not exist yet.

The condition alone stops a second count: a retry finds the like row, and the whole transaction is cancelled. Both writes happen, or neither does.

Add a ClientRequestToken, a unique id for the request. The docs say a retry with the same token within 10 minutes returns success without changing anything, instead of an error. Without a token, or after those 10 minutes, the retry is cancelled with the reason ConditionalCheckFailed. The service should treat that as already counted, not as an error.

This costs throughput. DynamoDB's docs say a transaction writes every item twice, once to prepare and once to commit. So each shard takes at most about 500 likes a second, not 1,000.

Transactions also conflict. DynamoDB cancels a transaction that touches a shard another transaction is still writing, with the reason TransactionConflict, and the capacity is still used. So 500 a second is a ceiling, not a plan.

Plan many more shards than that, and retry cancelled transactions after a short random wait, since the AWS SDKs do not retry them. The docs give no conflict rate, so load-test and count cancellations.

Store two numbers per reel: write_shards, the shards new likes go to, and read_shards, the shards readers add up. The one rule is that readers must never add up fewer shards than any writer uses.

Both numbers are cached on every likes server, because they are read on every like. That cache is where most resizing bugs hide.

• To grow from 1 to 20 shards, raise read_shards to 20 first. Wait until every server's cached settings have expired, then raise write_shards to 20.
• If you raise write_shards first, a server with the old cached value still adds up only shard 0. In a simulation, it showed 548 likes while the true total was 1,500.
• To shrink from 20 back to 1, lower write_shards to 1 first, then wait for the caches to expire. Only then fold each old shard into shard 0.
• Fold in one transaction: add the old shard's value to shard 0 and subtract it from the old shard together. Then lower read_shards and delete the empty shards.

Two separate writes are not safe for the fold. Add first, and a reader between the two writes counts those likes twice; subtract first, and it misses them. In the simulation, 140 likes briefly read as 180 or 100.

DynamoDB can do the fold with TransactWriteItems, which groups up to 100 actions. That keeps the stored total right, but a batch read can still catch the fold halfway.

DynamoDB's docs say a plain read of several items, such as BatchGetItem, can see some new and some old values while a transaction is still running. So the reader above can briefly show 180 or 100 too.

If even that blip matters, read the shards with TransactGetItems during a fold. It reads them as one unit, reads every item twice, and is cancelled while a write to them is running, so retry it.

The team accepts the blip, because the reel has already cooled down. Redis Cluster cannot do this fold. Its MULTI command, which runs several commands as one, fails when the keys sit in different slots, and spread-out shards do.

If your store has no multi-key transaction, skip the shrink. Stop writing to the extra shards and keep reading them, since reading 20 small items is cheap.

Four more traps catch teams after launch. Shards can land in the same place anyway, in two different ways. Shards can go negative, and reads can quietly become the new bottleneck.

The shards land in one Redis Cluster slot

Sharding only helps if the shard keys really hash to different places. Redis Cluster maps each key to one of 16,384 hash slots, and in a stable cluster each slot is served by one main node.

If a key contains a hash tag, a part inside curly brackets, only that part is hashed. Hash tags exist so related keys share one slot, which is the opposite of what shards need.

You can check this on your own machine with Redis 7. Start a server with redis-server --port 7000 --cluster-enabled yes, then ask it which slot each key maps to.

redis-cli -p 7000 CLUSTER KEYSLOT "{reel:8841}:likes:3"
redis-cli -p 7000 CLUSTER KEYSLOT "{reel:8841}:likes:19"
redis-cli -p 7000 CLUSTER KEYSLOT "reel:8841:likes:3"
redis-cli -p 7000 CLUSTER KEYSLOT "reel:8841:likes:19"

On Redis 7.0.15 the first two print 5809, the same slot. The last two print 7443 and 6464, different slots. A plain server without cluster mode refuses the command with an error.

Spread-out shards have one more cost in Redis Cluster. MGET reads many keys in one command, but only when they share a slot. An MGET of all 20 shards fails with a CROSSSLOT error, so the client sends 20 separate reads, ideally pipelined (sent together without waiting for each reply).

The shards sit side by side in a sorted store

Not every store hashes keys. Bigtable, a Google database that stores very large sorted tables, splits a table into tablets, blocks of contiguous rows sorted by key, and each tablet belongs to one node.

There, reel#8841#likes#0 to reel#8841#likes#19 sort next to each other and can share one tablet. Bigtable does split a tablet under sudden heavy load and move half to another node.

In a sorted store, put the shard number at the front instead, as in 07#reel#8841#likes. Keys that start differently are spread through the sorted key range, so the shards no longer sit together.

Unlikes and negative shards

When a fan removes a like, the service deletes the like row first, then subtracts 1 from a random shard. That shard can go below zero, which looks wrong but is fine, because only the sum matters.

Redis allows this: DECR, which subtracts 1, returns -1 on an empty key. So never show or check a single shard on its own.

Reads become the new bottleneck

If someone removes the cache, or sets it to zero seconds to look more live, every view sums every shard again. The team writes the cache rule into the design document, with the 1,000,000-reads maths beside it.

The main alternatives are batching likes in memory, sending like events through a log, or keeping one counter when traffic is low. Each one trades freshness, safety and effort differently.

Batching in memory works like this. Each likes server keeps its own running tally per reel and writes one +N every second. With 40 servers, a viral reel gets at most 40 writes a second, but a crashed server loses up to a second of likes.

A log-based design writes every like as an event to a log such as Kafka. A log is an append-only list of events that programs read in order. A consumer, a program that reads the log, adds up likes per reel and writes totals every few seconds.

It is replayable, but it has more parts to run. The Kafka design docs on Confluent's site warn about one failure. A consumer that crashes after processing events but before saving its position in the log may process some again.

These ideas also combine. A team could batch in memory and still write to sharded counters, for a reel so big that even 40 writes a second per key is too many. A sensible first step is sharded counters for hot keys only, because it is the smallest change.

Skip them when traffic per key is low, when you need an exact value on every read, or when the count guards a limit such as stock. In those cases, the extra reads and the drift cost more than they save.

• Low traffic: a reel with 5 likes a minute is fine with one counter.
• Exact reads: a wallet balance must be exact, so use transactions, not a cached sum.
• Limits: selling the last 10 seats needs a check and a write together, which a random shard cannot do safely.
• Unique counts: counting distinct viewers needs a set of ids, not a plus-one counter.

Explain it as a fix for one specific problem, a hot key, and walk from the failure to the fix to the read cost.

1. 1

   Show the hot key

   Draw one counter for a viral post. Say that all its writes hit one partition, so adding machines does not help.

2. 2

   Split the counter

   Replace it with N counters under keys like post#id#likes#0 to N-1. Each write adds 1 to a random one.

3. 3

   Size it with numbers

   Divide planned peak writes a second by a safe rate per key, below the cap. Say why: random choice is uneven.

4. 4

   Fix the read side

   Sum all shards in one batch read, and cache the total for a second or two. Show the read maths with and without the cache.

5. 5

   Shard only hot keys

   Store the shard count per key, start at one, and raise it when a monitor sees a key get hot. Raise the read count before the write count.

6. 6

   Handle retries and drift

   Say increments are not safe to retry. Keep a who-liked-what table as the truth, and run a job that corrects drift.