The Sharded Counter
In one line: the mechanism is almost trivially simple. What makes it a good design is the asymmetry it exploits — and being able to state that asymmetry is what separates understanding from recall.
The mechanism
When a user likes a tweet, the system forwards the request to the counter service. The service selects an available shard and increments it.
In the illustration, the total number of shards is (N+1). We select N based on the expected load.
Before: one counter After: N+1 shards
likes = 4,182,006 shard_0 = 597,441
shard_1 = 596,802
every write serializes shard_2 = 597,120
here ...
shard_N = 596,955
---------
total = 4,182,006 (computed on read)
A popular YouTube video may experience a burst of incoming views. At upload time, the system initializes a set of counter shards. For each view event, the system selects a shard — for example via random or hash-based routing — and increments that shard's counter. To compute the total view count, the system aggregates the values across all shards.
The design exploits an asymmetry — say it explicitly and everything follows
Sharding a counter does not make it faster in aggregate. It moves work from writes to reads:
- Writes go from "contend on one value" to "touch one of N values, uncontended." Contention drops by roughly N.
- Reads go from "read one value" to "read N values and sum them." Cost rises by roughly N.
That is only a good trade because of the workload's shape. For a viral tweet, writes and reads are both enormous — but writes are the ones that serialize. A million concurrent reads of one value proceed in parallel with no coordination at all; a million concurrent writes do not.
So the design converts an unparallelizable cost into a parallelizable one. Reads got more expensive but stayed embarrassingly parallel, and Lesson 6 makes them cheap again by aggregating periodically and caching the result.
The one-line version worth having ready: sharding trades read amplification for write parallelism, and that is a good trade precisely because only writes serialize.
Notice that shards are created at object creation time, not on demand
"At upload time, the system initializes a set of counter shards."
The shards exist before the first view arrives. That is deliberate, and it matters.
Creating shards lazily — on detecting load — means the burst has already started by the time you react, and bursts are exactly when you can least afford a control-plane operation. Lesson 4 notes that engagement "spikes shortly after publication", so the window between "content published" and "traffic arrives" is very short.
Pre-creating trades a small amount of waste (shards for content that never goes viral) for readiness. Given that a counter shard is a few bytes, that waste is negligible and the readiness is valuable.
The general principle: provision ahead of predictable bursts rather than reacting to them. Lesson 4 still adds dynamic resizing for the unpredictable cases, but the initial allocation is upfront.
'Each running in parallel on a different node' is the strong version
The source specifies shards on different nodes, not just different keys on one node — and that distinction determines how much you actually gain.
Shards as different rows in one database remove row-level lock contention but still share the database's CPU, disk, and network. Shards on different nodes remove all of it.
In practice you get to choose your level, and the right one depends on where the bottleneck is:
- Different keys, same store — removes lock contention. Often enough.
- Different partitions — removes partition-level hotspotting, which is message-queue design's hot-partition problem.
- Different nodes — removes machine-level saturation. Needed at the extreme.
For a tweet receiving millions of likes per minute, you want the third. Sharding is only as effective as the resource it separates.
What sharding gives up: you can no longer read the counter cheaply or exactly
Two things are lost, and both matter later:
Cheap reads. One value became N values. Lesson 6's answer is periodic aggregation into a cached total — which works, at the cost of freshness.
A single point of truth. There is no moment when "the counter" has a definite value, because the N shards are updated independently and read at slightly different times. The sum you compute is a value the counter passed through, not one it necessarily held.
For likes and views that is completely fine — nobody can tell whether a post has 4,182,006 or 4,182,011 likes. For anything requiring a read-then-write decision — inventory, balances, quotas — it is not, and Lesson 6 says so explicitly.
Sharded counters are for counting things, not for controlling things.
Key takeaway
One logical counter becomes N+1 physical shards; writes pick one, reads sum all. This does not increase aggregate throughput — it converts a serialized cost into a parallel one, which works because only writes require exclusive access. Shards are pre-created at object creation rather than on demand, and the design gives up cheap reads and a single point of truth.
What this pattern is actually called
Worth knowing the formal name, because an interviewer may use it and because it explains why the pattern is safe rather than just that it works.
A sharded counter is a G-Counter: a grow-only counter expressed as a conflict-free replicated data type. Each shard owns exactly one slot and only ever increments its own, so two shards can never disagree about a value — there is no shared cell to contend over.
The merge function is addition, and addition over disjoint slots is commutative and associative. That is the formal reason the design is safe: shards can be summed in any order, at any time, and the result converges without coordination.
It also explains the two limits the chapter runs into. Decrements break the structure — which is why a like-and-unlike counter needs a PN-Counter, a pair of grow-only counters where the value is increments minus decrements. And because the merge only ever grows, you cannot reset a shard without losing history, which is why resizing needs the care Lesson 4 describes.
Naming it connects this chapter to CRDTs generally: same convergence argument, much simpler data type.
Interview signal by level
| Level | What a strong answer sounds like |
|---|---|
| L4 | "Split the counter into several counters and add them up when reading." |
| L5 | Names the trade: "writes spread across N shards so contention drops by N, but reads now have to fetch and sum N values instead of one." |
| Staff+ | States the asymmetry and the loss: "this doesn't add aggregate throughput — it converts an unparallelizable cost into a parallelizable one, which only works because writes serialize and reads don't. What we give up is a single point of truth: the sum is a value the counter passed through, not one it definitely held. Fine for likes, wrong for anything read-then-write like inventory. And I'd pre-create shards at publication rather than reacting to load, because engagement spikes immediately and a control-plane operation mid-burst is the worst time for one." |
Next: the three operations.