How Would You Design a Distributed Counter (e.g. Like Counts)?
Learn how to design a distributed counter using sharding, cached aggregation, batching, and idempotency for high write throughput.
Expected Interview Answer
A distributed counter tracks a fast-changing count (like a like button) across many concurrent writers by sharding the counter into multiple sub-counters that absorb writes in parallel, then aggregating them for reads instead of contending on a single shared value.
A naive single-row counter with a database UPDATE ... SET count = count + 1 becomes a hot lock under high concurrency, since every increment serializes against the same row. The fix is to shard the counter into N sub-counters (e.g., in Redis or a sharded database), where each increment writes to one randomly or hash-selected shard, spreading contention across N rows instead of one. Reading the total sums all shards, which can be cached with a short TTL since exact real-time precision is rarely needed for a like count (eventual consistency is acceptable). For extreme write volume, increments can additionally be batched in memory or through a local aggregator before being flushed periodically, trading a small window of staleness for a large reduction in write amplification. Idempotency keys prevent double-counting when a client retries a failed increment request.
- Sharding spreads write contention across many rows instead of one hot lock
- Cached, periodically-aggregated reads avoid summing shards on every request
- Batching/buffering increments reduces write amplification under extreme load
- Idempotency keys prevent double-counting on client retries
AI Mentor Explanation
A distributed counter is like tallying runs scored across a stadium using several scorers at different vantage points instead of one scorer trying to record every ball alone. Each scorer (shard) tracks their own running subtotal independently, so no one scorer becomes a bottleneck during a flurry of boundaries. The scoreboard operator periodically adds up all scorers’ subtotals to display the official total, accepting it may lag by a ball or two rather than updating instantly on every single run. That parallel tallying with periodic aggregation is exactly how a distributed counter avoids a single point of contention.
Step-by-Step Explanation
Step 1
Shard the counter
Split the logical counter into N sub-counters (e.g., Redis keys or DB rows), spreading writes across shards.
Step 2
Route increments
Each increment writes to a hash- or randomly-selected shard, avoiding a single hot row under concurrency.
Step 3
Aggregate for reads
Sum all shard values on read, caching the total with a short TTL since exact real-time precision is rarely required.
Step 4
Batch under extreme load
Buffer increments locally and flush periodically for very high write rates, trading slight staleness for lower write amplification.
What Interviewer Expects
- Identifies the single hot-row contention problem with a naive counter
- Proposes sharding the counter and explains how reads aggregate shards
- Discusses caching the aggregate with acceptable staleness (eventual consistency)
- Mentions idempotency to prevent double-counting on retries
Common Mistakes
- Using a single row/key with SET count = count + 1 under high concurrency
- Summing all shards synchronously on every single read without caching
- Ignoring idempotency, causing retried increments to double-count
- Over-sharding to the point that aggregation itself becomes expensive
Best Answer (HR Friendly)
“A distributed counter splits a single fast-changing number, like a like count, into several smaller counters that can be updated in parallel, instead of one number everyone has to wait to update. We add up the smaller counters to show the total, and cache that total briefly since being off by a moment is fine for something like a like count.”
Code Example
const SHARD_COUNT = 16
async function increment(counterId) {
const shard = Math.floor(Math.random() * SHARD_COUNT)
await redis.incr(`counter:${counterId}:${shard}`)
}
async function getTotal(counterId) {
const cached = await cache.get(`counter:${counterId}:total`)
if (cached !== null) return cached
const shardKeys = Array.from(
{ length: SHARD_COUNT },
(_, i) => `counter:${counterId}:${i}`,
)
const values = await redis.mget(shardKeys)
const total = values.reduce((sum, v) => sum + (parseInt(v, 10) || 0), 0)
await cache.set(`counter:${counterId}:total`, total, { ttlSeconds: 2 })
return total
}Follow-up Questions
- How would you decide the right number of shards for a given write throughput?
- How would you make increments idempotent when a client retries a failed request?
- What consistency model is acceptable for a like counter versus an inventory counter, and why?
- How would you handle decrementing (e.g., unlike) without double-processing under retries?
MCQ Practice
1. What problem does sharding a counter primarily solve?
Sharding distributes concurrent increments across N sub-counters, avoiding serialization on a single contended row.
2. Why is a short-TTL cache commonly used for the aggregate read of a distributed counter?
Many counters like likes tolerate brief staleness, so caching the summed total avoids recomputing it on every single read.
3. Why are idempotency keys important for a distributed counter?
Without idempotency, a client retry after a timeout could apply the same increment twice, inflating the count.
Flash Cards
Why does a naive single-row counter fail at scale? — Every increment serializes on the same row/lock, creating a bottleneck under concurrency.
How does sharding fix counter contention? — Splitting into N sub-counters spreads writes across N rows instead of one.
Why cache the aggregated total? — Summing all shards on every read is wasteful; a short-TTL cache trades minor staleness for efficiency.
Why use idempotency keys on increments? — To prevent a retried request from double-counting the same increment.