Distributed Counter — Sharded Aggregation

Counting things at scale sounds trivial until you realize that a single viral video can receive millions of view increments per second. The distributed counter problem appears in nearly every large-scale consumer application: YouTube view counts, Instagram likes, Twitter retweet counts, and click-through analytics. It is a popular system design interview question because it tests a candidate's ability to reason about write-heavy workloads, hot keys, eventual consistency, and the trade-offs between accuracy and performance.

The fundamental challenge is that a single counter key (such as a viral video's view count) can become an extreme hot spot. A single Redis key receiving 10 million INCR operations per second would saturate a single Redis core, which tops out at roughly 1 million operations per second. The naive approach of incrementing a single database row is even worse, as it creates lock contention and write amplification that can bring down the entire database tier.

The sharded counter pattern solves this by splitting each logical counter into N sub-counters (shards). Write operations increment a randomly selected shard, distributing the load across multiple Redis hash slots and potentially multiple cluster nodes. A background aggregation worker periodically sums all shards to produce an approximate total for fast reads. The trade-off is that the displayed count is slightly stale (typically by a few seconds), which is perfectly acceptable for social media metrics but would not work for financial counters.

At production scale, systems like YouTube handle billions of counter updates per day with peak bursts during viral events. The system must sustain 1 million increments per second during normal operation and absorb 10 million per second viral spikes without dropping events. Read latency must remain under 5 milliseconds for the approximate count that powers the user interface.

The architecture cleanly separates the write path (optimized for throughput) from the read path (optimized for latency), with an asynchronous aggregation layer bridging the two. This separation allows each path to scale independently based on its unique workload characteristics.

On the write path, counter increment requests flow through the API Gateway (JWT authentication, per-user rate limiting to prevent bot abuse) and load balancer to the CounterService. CounterService hashes the item ID with a random shard index to select one of 16 sub-counter shards, then performs an atomic INCR on ShardCache (a 12-node Redis cluster). The Redis INCR completes in approximately 2ms. CounterService also publishes a fire-and-forget increment event to AggStream (Kafka with 128 partitions), which buffers up to 5 million messages to absorb viral traffic spikes. No database is touched on the hot write path.

The aggregation layer consists of two worker types. The AggregatorWorker runs continuously with 40 worker instances, consuming from AggStream. Every 5 seconds per active key, it reads all 16 sub-counter shards via MGET from ShardCache, computes the sum, and writes the aggregated total to MasterCache (a separate 6-node Redis cluster). This produces the approximate count that is at most 5 seconds stale. The PersistWorker batches aggregated counts and writes them to CounterDB (DynamoDB on-demand) every 30 seconds for durable storage, reducing write costs by 30x compared to per-increment database writes.

On the read path, ReadService handles two endpoints: the approximate count (GET from MasterCache, approximately 2ms, used by 95% of reads) and the exact count (query from CounterDB, approximately 15ms, used for analytics and auditing). The separation of ShardCache and MasterCache prevents write amplification from the high-throughput shard INCR operations from interfering with read latency on the master cache.