Hot keys and the celebrity problem: an interactive shard load simulator

Sharding spreads keys evenly across machines, but not traffic. A hot key, such as a celebrity’s profile, a viral post’s like counter or a flash sale’s inventory, sends a large share of all requests to the one shard that owns it, which overloads while the rest sit idle. Adding shards does not help, because the key still lives on one of them. The fix depends on whether the heat is reads or writes.

The playground sends 50,000 operations a second to 8 shards of 10,000 each and points a share of them at one key. Change how hot it is and how many of its operations are writes, and compare five strategies.

Reads: copy the key

When 30% of traffic reads one key, its shard runs at 194% of capacity while the other seven idle at 44%. Reads can be answered by copies. A cache in each app server, refreshed every second, turns 14,250 reads a second into 50 refreshes and brings the shard down to 52%, at the price of reads up to a second stale. Replicating the key to 4 shards spreads the reads, 87%. Splitting the key is the wrong tool here: each read has to fetch all 8 pieces, and every shard ends up at 187%.

Writes: split the key

When the same traffic is 95% writes, a cache does nothing, 187%, and replicas make it worse, since every copy must apply every write. Writes need the key split into sub-keys on different shards: a write picks one sub-key at random, and a read adds them all up. For a like counter that brings the busiest shard to 69%. With heavy reads and heavy writes together, no single fix is enough, and the answer is both: split the key for writes and cache its total for reads, 54% here.

• Find hot keys before they find you. Sample the keys each shard serves, or track the top keys with a count-min sketch, and alert on any key taking a large share.
• Plan for known ones. Celebrities, big customers and launch-day products are predictable, and can be split or cached in advance.
• Buffer writes. For counters, aggregate increments in memory or a queue and write a total every second, a write-back approach.
• Protect the shard anyway. Per-key rate limits and request coalescing stop one key from taking the shard down for every other key on it.

A split counter with a cached total

import random, time
import redis

r = redis.Redis()          # a Redis Cluster in production: each sub-key hashes to its own slot
SPLIT = 8
_cache = {}                # per app server: key -> (value, fetched_at)

def like(post_id):
    # spread writes over SPLIT sub-keys, so no single shard takes them all
    r.incr(f"likes:{post_id}:{random.randrange(SPLIT)}")

def like_count(post_id, ttl=1.0):
    hit = _cache.get(post_id)
    if hit and time.monotonic() - hit[1] < ttl:
        return hit[0]                      # most reads never leave the app server
    keys = [f"likes:{post_id}:{i}" for i in range(SPLIT)]
    total = sum(int(v or 0) for v in r.mget(keys))   # one read of every piece
    _cache[post_id] = (total, time.monotonic())
    return total

In a system design interview

Bring up hot keys yourself whenever traffic follows popularity: “Most posts get a few likes, but a viral one can take thousands a second, all on one shard. So a post’s like counter is split into 16 sub-keys once it crosses a threshold, writes go to a random sub-key, and the displayed count is the sum, cached for a second. Reads of hot posts are served from the CDN and app caches.” Expect the follow-ups: how you detect a key turning hot, how reads of a split counter stay cheap, and what happens to ordering and exact counts.

Read the guide to hot keys and skew or counting at scale, see choosing a shard key, or watch a strong candidate design a news feed.

Questions people ask