Hot keys and hot partitions

Why one key can melt a shard, how to detect it, and how to fix it by caching, replicating, salting, batching or shedding, with numbers for interviews.

Senior33 min core · 34 min full depth

Read first: Partitioning Caching

A hot key is one key that receives far more traffic than its share. Examples are one celebrity’s profile, one viral post’s like counter, one concert’s seat inventory, or one tenant’s rate-limit bucket. A partitioned system routes every request for a key to the one shard that owns it. So a hot key becomes a hot partition: one shard at 100%, while the rest of the cluster idles. Adding shards does not fix it. The fixes act on the key itself: detect it, split it, cache it, or shed it.

This page covers the pattern: where skew comes from, how to find hot keys, which fix fits which traffic, and what each costs. The mechanisms (salting, stampede protection, heavy-hitter sketches) live on the building block pages linked below.

The problem it solves

A Redis cluster has 8 shards, each good for about 150,000 operations a second, so 1.2 million in total. Traffic is 600,000 a second, half of capacity, and every dashboard is green. Then a footballer with 600 million followers posts a photo. Within a minute, one key, the post’s metadata, takes 200,000 reads a second, and all of them hash to the same shard. That shard was carrying its eighth of normal traffic, about 75,000 a second, and now carries 275,000: over 180% of what it can serve. Its queue grows without bound, every request on it times out, and every other key on that shard fails with the celebrity’s. The other seven shards sit at 50%.

Clients retry the timeouts, adding load where there is none to spare. If the replica is promoted, it receives the same key and dies the same way. And the cluster’s total load, the number capacity planning watches, is only 67%.

Where skew comes from

Uniform traffic is the exception. Five shapes produce hot keys in practice:

Source	Example	Shape
Natural popularity	Product pages, videos, cache keys in general	Zipf: steady, predictable, a long tail
Celebrities and whales	An account with 100 million followers, a tenant 1,000 times larger than the median	Known keys, permanently hot
Scheduled spikes	A flash sale, a concert on-sale, a Super Bowl ad	Known in advance, hot for minutes
Viral content	A post, a link, a search term that takes off	Unknown key, cold to hot in seconds
Time-ordered keys	Auto-increment IDs, timestamps as the key in a range-partitioned store	Every new write goes to the last range

A sixth source is a design bug: a key that every request touches. Examples are one global configuration blob, or a rate-limit rule keyed by its name instead of by the caller. The fix for those is a different key.

Zipf’s law is the shape behind most of them. The item of rank r gets traffic in proportion to 1 / r^s. For many web workloads, s is near 1. (Cache traces often measure 0.6 to 0.9, and social content measures more.) With s = 1 over a million keys, the most popular key gets 7% of all traffic. The top 100 keys get 36%, and the top 1% of keys get 68%.

That 7% is the number that matters. Say each shard serves C operations a second. A key with share p puts p × total on one shard. So the cluster saturates at total = C ÷ p, however many shards it has. At C = 150,000 and p = 7%, the ceiling is about 2.1 million operations a second, with 8 shards or with 800.

Why more shards make it relatively worse

Hashing evens out how many keys each shard holds, not how much traffic one key brings. More shards lower each shard’s fair share while the hot key’s load stays put:

Shards	Fair share of 800,000 ops/s	A 7% key brings	Hot key ÷ fair share
8	100,000	56,000	0.56
32	25,000	56,000	2.2
128	6,250	56,000	9

Scaling out lowers the mean and leaves the max where it was.

What runs out first

“Hot” means one key exhausts one resource on one machine:

CPU on a single-threaded engine. One Redis primary runs commands on one thread, about 100,000 to 200,000 simple operations a second.
Network. A 1 KB value read a million times a second is about 8 Gbit/s. That’s most of a 10 Gbit/s card, before the CPU is even busy.
Row locks. UPDATE posts SET likes = likes + 1 waits in line on the row lock. So one row takes on the order of a thousand updates a second, however large the machine is.
Hard per-partition limits. A DynamoDB partition serves at most 3,000 read units and 1,000 write units a second. Firestore warns that one document cannot be updated at an unlimited rate and plans its sharded counters around roughly one write a second per shard. A Kafka partition is read by one consumer in a group.

One distinction decides everything after this: is the key read-hot or write-hot? The fixes for one are close to the opposite of the fixes for the other.

How it works

Step 1: Find the hot keys

A per-shard dashboard shows that shard 3 is at 100%. It doesn’t show which key did it. Every fix below applies to specific keys, so detection comes first. And it must answer within seconds: a viral key goes from nothing to its peak in under a minute. Three places can count:

On the shard. Exact per-key counts cost memory and CPU on the machine that is already overloaded. Redis can list its hottest keys (redis-cli --hotkeys, with an LFU eviction policy), which helps diagnosis and is too slow for automatic reaction.
In a sample. Count 1 request in 100 into a bounded top-k structure. That can be Space-Saving (a fixed table of k counters), or a Count-Min sketch plus a heap. Both are on the probabilistic structures page. A key at 10,000 a second still yields 100 samples a second.
On the client. Each app server or sidecar samples its own requests. It sees a hot key even when the shard is too busy to report, and most fixes are applied there anyway.

The usual design samples on the clients, merges every second, and publishes a hot list that every client reads:

The hot-key loop: sample, merge, publish, act

Scroll or pinch to zoom, drag to pan

Color key
Database
Background worker
Service

The detector below samples 1 in 100 of a million requests. That’s one second of traffic over 100,000 keys, Zipf-shaped, with one viral key at 20%. It keeps 256 Space-Saving counters, and flags keys above 10,000 a second:

HotKeyDetector.javahot_key_detector.pyHotKeyDetector.cshot_key_detector.gohot-key-detector.ts

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;

// A tiny seeded generator, written the same way in all five languages so they print the same
// result. Any decent random source works in production.
class XorShift32 {
    private int x;

    XorShift32(long seed) { x = (int) seed; }

    long next() {
        x ^= x << 13;
        x ^= x >>> 17;
        x ^= x << 5;
        return x & 0xFFFFFFFFL;
    }

    double uniform() { return (next() >>> 8) / 16777216.0; }
}

// A hot-key detector: sample 1 in N requests into a Space-Saving top-k, report per window.
class HotKeyDetector {
    record Hot(String key, double rate, double atLeast) {}

    private final int sampleEvery;
    private final int capacity;
    private final double threshold;
    private final Map<String, long[]> counts = new HashMap<>(); // key -> {count, error}
    private final XorShift32 rng; // its own random source, independent of the traffic

    // Sampling keeps the cost off the request path: 1 in 100 requests touches the table.
    // A key at 10,000 a second still yields about 100 samples a second, plenty to see it.
    // Space-Saving keeps at most `capacity` counters, whatever the number of distinct keys.
    HotKeyDetector(int sampleEvery, int capacity, double thresholdPerSec, long seed) {
        this.sampleEvery = sampleEvery;
        this.capacity = capacity;
        this.threshold = thresholdPerSec;
        this.rng = new XorShift32(seed);
    }

    void observe(String key) {
        if (rng.next() % sampleEvery != 0) return;
        long[] entry = counts.get(key);
        if (entry != null) {
            entry[0]++;
        } else if (counts.size() < capacity) {
            counts.put(key, new long[] {1, 0});
        } else {
            // The table is full: the new key replaces the smallest counter and inherits its
            // count as possible error. A key that is truly hot is never the smallest for long.
            // Ties break on the key so the eviction is the same in every language.
            String victim = null;
            for (var e : counts.entrySet()) {
                if (victim == null) { victim = e.getKey(); continue; }
                long c = e.getValue()[0], v = counts.get(victim)[0];
                if (c < v || (c == v && e.getKey().compareTo(victim) < 0)) victim = e.getKey();
            }
            long floor = counts.remove(victim)[0];
            counts.put(key, new long[] {floor + 1, floor});
        }
    }

    // Scale samples back to a rate. A counter can overstate a key by its inherited error, so
    // the decision uses count - error, a lower bound: a key evicted and re-added many times
    // in the long tail never looks hot because it inherited someone else's count.
    List<Hot> flush(double windowSec) {
        List<Map.Entry<String, long[]>> entries = new ArrayList<>(counts.entrySet());
        entries.sort(Comparator.<Map.Entry<String, long[]>>comparingLong(e -> -e.getValue()[0])
                .thenComparing(Map.Entry::getKey));
        List<Hot> report = new ArrayList<>();
        for (var e : entries) {
            long count = e.getValue()[0], error = e.getValue()[1];
            double floor = (count - error) * sampleEvery / windowSec;
            if (floor >= threshold) report.add(new Hot(e.getKey(), count * sampleEvery / windowSec, floor));
        }
        counts.clear(); // a fresh table per window, so a key that cools off drops out
        return report;
    }
}

class Main {
    public static void main(String[] args) {
        int keys = 100_000;
        double s = 1.0;
        double[] cdf = new double[keys];
        double total = 0;
        for (int r = 1; r <= keys; r++) { // Zipf: the key of rank r gets traffic in proportion to 1/r^s
            total += 1.0 / Math.pow(r, s);
            cdf[r - 1] = total;
        }

        XorShift32 rng = new XorShift32(2463534242L);
        HotKeyDetector detector = new HotKeyDetector(100, 256, 10_000, 88675123L);
        Map<String, Long> truth = new HashMap<>();
        int requests = 1_000_000; // one second of traffic
        for (int i = 0; i < requests; i++) {
            String key;
            if (rng.uniform() < 0.2) {
                key = "video:viral"; // one key suddenly takes 20% of everything
            } else {
                int idx = Arrays.binarySearch(cdf, rng.uniform() * total);
                key = "video:" + ((idx >= 0 ? idx : -idx - 1) + 1);
            }
            truth.merge(key, 1L, Long::sum);
            detector.observe(key);
        }

        for (var h : detector.flush(1.0)) {
            System.out.println(String.format(Locale.ROOT, "%-12s estimated %,9.0f/s  at least %,9.0f/s  true %,9d/s",
                    h.key(), h.rate(), h.atLeast(), truth.get(h.key())));
        }
    }
}

The viral key and the top five come out within about 7% of their true rates, from 10,000 samples and 256 counters. Key 6, truly at about 11,000 a second, sits just over the threshold. It is missed in this window: near the line, sampling noise decides.

So real detectors use two thresholds with a gap between them. A key is promoted above the upper one. It’s demoted only after several windows well below it. Then a key near the line does not flap between treatments. (This gap is called hysteresis.)

Space-Saving guarantees a counter for any key with more than samples ÷ k of the window. Here 10,000 ÷ 256 is 39 samples, or about 3,900 a second. That’s well under the threshold, so no hot key can be crowded out.

There’s one catch. A key that evicts the smallest counter inherits that count as possible error. So a cold key can look warm. The decision therefore uses count − error, a lower bound. It never promotes a key on a count it inherited. The price is that a key that arrived late in the window is promoted one window later. Promoting a cold key is not free, since every fix below has a cost.

Step 2: Read-hot keys: absorb reads before the shard

A read-hot key has one value and many readers. The fixes put copies of the value closer to the readers, so most reads never reach the shard.

A local in-process cache with a short TTL. Each app server keeps hot keys in memory for 1 to 5 seconds. The shard then sees at most one read per app server per TTL for that key, whatever the traffic. 200 app servers with a 1 second TTL is 200 reads a second, down from 200,000. The price is staleness up to the TTL, and memory. That’s why the local cache holds only detected hot keys, or uses a frequency-aware policy such as W-TinyLFU.

Request coalescing (single flight). When a local entry expires, the first request reloads it and the others wait. Without it, every request in the reload window goes to the shard, every TTL. The caching page has the mechanism and a demo.

Replicate the key to N cache nodes. Store the value under post:9#0 to post:9#7, which hash to different shards. Each reader picks a suffix at random. Read load per shard falls by N. But every write and invalidation now goes to all N copies. And for a moment, copy 3 can be updated while copy 5 is not. Reading from a shard’s replicas is the same idea, with a lower ceiling. A primary has only a few replicas, and they lag.

A CDN. If the response is the same for every viewer, the edge serves it. It collapses concurrent misses into one origin request. And with stale-while-revalidate, it serves the old copy during a refresh. See CDN.

Layers that absorb reads of one hot key

Scroll or pinch to zoom, drag to pan

Color key
Service
Database

The layers multiply: 200 reloads a second reach the cache tier, and with 8 copies each copy sees 25. Replication earns its place in very large fleets. The distributed cache problem has 20,000 hosts, where even one read per process every 2 seconds is 160,000 reads a second on one shard.

Step 3: Write-hot keys: spread the writes, or make fewer of them

A write-hot key cannot be cached away: every like, view or click must land somewhere. Either spread one key’s writes over many keys, or combine many writes into one.

Salting, also called a sharded counter. One logical counter is stored as N slot keys, likes:post9#0 to likes:post9#31. A write increments one slot, and a read sums them all. The slot comes from a hash of the event ID, not a random number. So a retried event lands on the same slot, and dedup and replay still work. The partitioning page shows the basic version. The production version adds three things:

Slots only for keys the detector flags.
A slot count that only grows.
A cached sum, so reads do not pay the fan-in on every request.

Store.javasharded_counter.pyStore.cssharded_counter.gosharded-counter.ts

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

// Stands in for a sharded key-value store, counting the operations each shard serves.
class Store {
    final List<Map<String, Long>> data = new ArrayList<>();
    final long[] writes;
    final long[] reads;

    Store(int shards) {
        for (int s = 0; s < shards; s++) data.add(new HashMap<>());
        writes = new long[shards];
        reads = new long[shards];
    }

    // A stable hash, the same on every server. String.hashCode() is stable but poorly mixed.
    static long fnv1a(String s) {
        long h = 0x811C9DC5L;
        for (byte b : s.getBytes(StandardCharsets.UTF_8)) {
            h ^= (b & 0xFF);
            h = (h * 0x01000193L) & 0xFFFFFFFFL;
        }
        return h;
    }

    private int shard(String key) { return (int) (fnv1a(key) % data.size()); }

    void incr(String key) {
        int s = shard(key);
        data.get(s).merge(key, 1L, Long::sum);
        writes[s]++;
    }

    // One batched read (MGET); each key still costs a read on the shard that owns it.
    long[] getMany(List<String> keys) {
        long[] out = new long[keys.size()];
        for (int i = 0; i < keys.size(); i++) {
            int s = shard(keys.get(i));
            reads[s]++;
            out[i] = data.get(s).getOrDefault(keys.get(i), 0L);
        }
        return out;
    }
}

// A sharded counter: one logical counter stored as N slot keys, summed on read.
class ShardedCounter {
    private record Cached(long value, double readAt) {}

    private final Store store;
    private final double readTtl;
    // Slot count per key, pushed by the hot-key detector. Absent means 1: a cold key costs
    // one key and one read, so only hot keys pay the fan-in.
    private final Map<String, Integer> slots = new HashMap<>();
    private final Map<String, Cached> cache = new HashMap<>();

    ShardedCounter(Store store, double readTtl) {
        this.store = store;
        this.readTtl = readTtl;
    }

    // Only ever grow. Readers sum slots 0..n-1, so shrinking would hide counts in the high
    // slots until they were folded back into slot 0.
    void promote(String key, int n) { slots.merge(key, n, Math::max); }

    // The slot comes from the event id, not a random number: a retried event lands on the
    // same slot, so per-slot dedup and replays still work. Hashing the event (not the
    // writer) spreads the load even when one writer sends everything.
    void incr(String key, String eventId) {
        int n = slots.getOrDefault(key, 1);
        store.incr(key + "#" + (Store.fnv1a(eventId) % n));
    }

    // Read aggregation: sum every slot in one batched read, then serve that sum from memory
    // for readTtl seconds. A hot counter costs n slot reads per second per process,
    // whatever the read rate.
    long read(String key, double now) {
        Cached hit = cache.get(key);
        if (hit != null && now - hit.readAt() < readTtl) return hit.value();
        int n = slots.getOrDefault(key, 1);
        List<String> keys = new ArrayList<>();
        for (int i = 0; i < n; i++) keys.add(key + "#" + i);
        long total = Arrays.stream(store.getMany(keys)).sum();
        cache.put(key, new Cached(total, now));
        return total;
    }
}

class Main {
    public static void main(String[] args) {
        Store store = new Store(8);
        ShardedCounter likes = new ShardedCounter(store, 1.0);
        String key = "post:42:likes";

        for (int i = 0; i < 20_000; i++) likes.incr(key, "like:" + i); // one slot, one shard
        System.out.println("1 slot:   writes per shard " + Arrays.toString(store.writes));

        likes.promote(key, 32); // the detector flags the key; writers learn the new slot count
        long[] before = store.writes.clone();
        for (int i = 20_000; i < 200_000; i++) likes.incr(key, "like:" + i);
        long[] after = new long[before.length];
        for (int s = 0; s < after.length; s++) after[s] = store.writes[s] - before[s];
        System.out.println("32 slots: writes per shard " + Arrays.toString(after));

        // 100,000 reads spread over 10 seconds, on one process.
        for (int i = 0; i < 100_000; i++) likes.read(key, i * 0.0001);
        System.out.println("100000 reads cost " + Arrays.stream(store.reads).sum() + " slot reads");
        System.out.println("count = " + likes.read(key, 60.0));
    }
}

Before detection, the first 20,000 likes all land on shard 0. After promotion to 32 slots, 180,000 likes spread over all 8 shards, about 17,000 to 28,000 each. (32 hashed slots do not split perfectly over 8 shards.) And 100,000 reads over 10 seconds cost 320 slot reads. That’s one batched read of 32 slots per second, because the sum is cached for a second.

The slot count must reach readers before writers. A reader that sums slots 0 to 7 while a writer uses slot 20 undercounts. Store the count beside the counter, or have every server learn it from the hot list before any server writes the new slots. Never shrinking is what keeps this safe.

Choosing N is arithmetic: the peak write rate divided by what one key can take, with headroom. The Instagram problem takes 20,000 likes a second on one post over 64 slots, about 300 a second per slot.

Buffering and batching. Each app server sums increments per key in memory. It flushes one INCRBY likes:post9 4400 a second. The key’s write rate on the store becomes the number of app servers, whatever the event rate. The top-k problem turns 525,000 views a second of one video into about 120 messages this way. A crash loses up to one flush of increments. So when they must not be lost, events go to a durable log, and a stream job sums them.

CRDT counters. A grow-only counter (G-counter) keeps one slot per replica or region. Each replica increments only its own slot. A merge takes the maximum of each slot, and the value is the sum. A PN-counter pairs two of them, for decrements. It is salting where the slot is “which region”, for a counter written in several regions with no coordination. Redis Enterprise Active-Active and Riak ship counters built this way.

Approximate counts. A view counter shown as “1.2M” does not need every increment. Increment with probability 1/k, by k. After a million events with k = 100, the standard error is about 10,000, or 1%. And the key takes a hundredth of the writes, with no fan-in at all.

Step 4: Hot rows that must stay exact: flash sales and inventory

A like counter can be approximate for a few seconds. A count of 10,000 sneakers cannot: the 10,001st sale is a refund and an apology. The write is a conditional decrement, and everyone wants it in the same second:

UPDATE stock SET available = available - 1
WHERE sku = 'AJ1-CHI-42' AND available >= 1;

On one row this serializes on the lock, about a thousand a second, against 50,000 people pressing “buy” at 10:00:00. Three fixes, usually combined:

Pre-split the stock into buckets. 16 rows of 625 units instead of one row of 10,000. A purchase decrements a random bucket and tries the next if it is empty. Contention falls 16 times and the total stays exact, because each unit lives in exactly one bucket. Near the end most buckets are empty and requests probe several, so the last units are slower unless a job rebalances buckets.
Queue-based admission. A waiting room lets people through at the rate the rows can take, say 500 a second. Everyone else waits, with a position. The Ticketmaster problem builds one for an on-sale. Without it, the sale would send 25,000 hold attempts a second to one event.
Token allocation. Before the sale, mint one token per unit into Redis lists. A buyer is admitted by popping one (LPOP, atomic and O(1)), and only token holders reach the database. Empty lists mean sold out, answered with no database work. Abandoned checkouts return their token after a timeout.

A flash sale with admission, tokens and bucketed stock

Scroll or pinch to zoom, drag to pan

Color key
Service
Database

Step 5: Shed what is left

Some hot traffic is not worth serving. It could be a client retrying 300,000 times a second, a scraper on one listing, or a key whose load exceeds every fix. A per-key rate limit at the gateway caps what one key can cost the store (see rate limiting). A view counter can freeze under overload, and show its last value. A hot tenant can be moved to a shard of its own. That isolates it, rather than fixing it. Shedding trades correct answers for availability, so it is the last layer.

Try it

Eight shards, each good for 150,000 operations a second, receive Zipf traffic over 100,000 keys. The detector flags any key above 30,000 a second, and the fixes apply only to flagged keys. Bars show each shard’s load against its capacity (the dashed line), with hot-key traffic in orange. The p99 comes from a simple queueing model in which latency climbs steeply as a shard nears 100%.

Traffic shape

Mix

Shards

Total traffic 800,000 ops/s

Fixes for detected hot keys

Hot-key trafficEverything elseShard capacity, 150,000 ops/s

p99 latency: -
Requests timing out: -
Busiest shard vs mean: -
Shard ops per request: -
Cluster ceiling at this shape: -

Things to try:

Start at Zipf 1.0 and 800,000 a second. Shard 1 holds key:1 at 66,000 a second. It runs past capacity, while the average shard is at 67%. About a fifth of requests time out. The cluster ceiling at this shape is about 800,000. So a 1.2 million ops/s cluster cannot take one more request of this mix.
Switch to Uniform. Every shard sits at 67% and p99 is about 3 ms. Same total, same hash. Only the shape changed.
Choose Zipf + viral key and 32 shards. Every other bar fell. But the viral key’s shard is still at about 140%, nearly nine times the mean.
With 95% reads, turn on salting alone. Every shard goes past capacity: each read of a salted key asks all 8 sub-keys. Swap salting for the local cache and the hot keys vanish from the bars.
Switch to 90% writes. Now the local cache barely helps. Replication melts every shard, because each write goes to all 8 copies. Salting brings p99 back. Then set the viral key at 50/50, with the local cache plus salting. That’s the pair most real hot counters end up with, and it keeps every shard healthy.

Options and tradeoffs

Fixes for read-hot keys

Fix	Load on the hot shard	Staleness	Extra cost	Wins when
Local cache, short TTL	App servers ÷ TTL, whatever the traffic	Up to the TTL (1 to 5 s)	Memory on every server; deletes wait a TTL	Almost always the first fix
Request coalescing	One load per server per expiry	None added	Waiters queue behind the first load	Always, alongside any cache
Replicate to N keys	Divided by N	Copies can briefly disagree	N writes and N invalidations per update	Fleets so large the local cache still overloads a shard
Read replicas of the shard	Divided by replicas (usually 2 to 5)	Replication lag	A ceiling of a few replicas per primary	A modest hot spot in a store that already has replicas
CDN	Near zero for cacheable responses	Up to `max-age`	Only public, identical responses	Media, link redirects, short search prefixes

Fixes for write-hot keys

Fix	Writes on the hot shard	Read cost	Exactness	Wins when
Salting / sharded counter	Divided by N	Sum N slots (cache the sum)	Exact	Counters and append streams at any rate
Buffer and batch	App servers per flush interval	One key	Exact but lagging; a crash loses a flush	Counters that tolerate a second of lag
Stream aggregation	One write per key per window	One key	Exact, replayable from the log	Counts that must not lose events: billing, ads
CRDT counter	One slot per region	Sum the slots	Converges	One counter written in several regions
Approximate count	Divided by the sampling factor	One key	Approximate	Display counts: views, likes in the millions
Bucketed stock + admission	Bounded by the admission rate	Sum buckets for “how many left”	Exact	Inventory and seats, which must never oversell

The alternative is to salt every key. The ad click aggregator salts every ad 16 ways. A cold ad’s empty partials cost nothing, and no one has to predict which ad goes viral. That works when reads are rare and batched, as in a stream that merges partials once per window. It is wrong when reads are frequent and individual.

Detection

Approach	Sees	Cost	Lag	Wins when
Per-shard metrics	Which shard, not which key	Free	Seconds	Alerting that something is wrong
Sampled top-k on clients	Keys above a rate, approximately	One random draw per request	1 to 3 s	Promoting fixes automatically
Store-provided hot-key reports	The store’s own view	Free to enable	Often minutes	Diagnosis: `redis-cli --hotkeys`, DynamoDB Contributor Insights
Known lists	Celebrities, scheduled events	A config file	None	Keys you know will be hot before they are

What the store does on its own

Managed stores rebalance, and it is tempting to assume that covers hot keys. It covers hot partitions holding many keys, never one key.

Store	What it does automatically	What it cannot do
DynamoDB adaptive capacity	Lets a hot partition use throughput others leave idle, up to the partition maximum	Exceed 3,000 reads or 1,000 writes a second on one partition
DynamoDB split for heat	Splits a hot partition so its hot items land on separate partitions	Split one item, help an increasing sort key, or split item collections on a table with an LSI
Bigtable, CockroachDB, Spanner	Split ranges by size and by load	Split below one row key
Redis Cluster	Nothing automatic; an operator moves slots between nodes	Split one of its 16,384 slots, or serve a key from more than its primary and replicas
Kafka	Nothing; a key always maps to one partition	Give one key more than one consumer’s throughput while keeping its order

Go deeperWhy time-ordered keys defeat load-based splitting1 min

In a range-partitioned store, think of a key that grows with time: a timestamp, an auto-increment ID, or a ULID as the primary key. It sends every new write to the last range. A load-based split divides that range in two. Then every new write goes to the upper half, which is hot again. The split moved the past, not the future.

The fixes change the key:

Prefix it with a hash bucket: HBase-style salting, with hash(id) mod 16 in front.
Reverse the bits of a sequence. (Spanner offers bit-reversed sequences for this reason.)
Lead with a high-cardinality field such as the user ID, and keep time second.

Each one turns “the latest data” into a fan-out over the buckets. That’s the same trade salting makes.

How it fails

One shard melts, and the failure spreads. Every key on the shard times out, and retries pile on. A promoted replica receives the same key, and melts. A rebalancer that moves the hot slot melts the next node. Alert on one shard far above the mean, not on total load. Fix the key, and give clients retry budgets.

A stampede on the hot key. When a hot key’s entry expires, the misses in the reload window scale with the key’s traffic. The caching page covers the fixes: single flight, leases and early refresh.

Fan-in that costs more than the hotspot. An uncached read of a salted key waits for the slowest of N shards. If 1% of shard reads are slow, a read over 8 slots hits one 8% of the time. Over 64 slots, it’s 47% of the time. Salt only write-hot keys, cache the sum, and keep N as small as the write rate allows.

Late or flapping detection. A 10 second window plus a 5 second publish delay lets the shard melt before the fix arrives. Without hysteresis a key flips treatments every window. A detector on the overloaded shard goes quiet when it is needed. Sample on clients, use 1 second windows, and expire hot lists so a dead aggregator cannot leave a stale one in force.

Copies that disagree, and deletes that wait. A writer crashes after updating 5 of 8 copies. Then readers see two values at random, until the next write. A takedown sits in every local cache for a TTL. Version the copies, retry invalidations, and push deletes to local caches.

Failure	Blast radius	Detect	Fix
One shard melts	Every key on the shard, then its replica	Per-shard load vs the mean; top keys per shard	Act on the key; retry budgets; never move the hot slot
Stampede on expiry	The database behind the hot key	Miss spikes aligned to one key’s expiry	Single flight, leases, early refresh
Fan-in cost of salting	Every shard, every read of the key	Shard ops per request rising; read p99 tracking N	Salt only write-hot keys; cache the sum; smaller N
Late or flapping detection	The hot shard, for the detection lag	Time from shard skew to promotion; flips per key	Client sampling, 1 s windows, hysteresis, expiring lists
Copies diverge	Readers of that key	Version mismatch across copies	Versioned writes, retried invalidation, short copy TTL
Stale local copy after delete	Viewers of that key for one TTL	Takedown-to-gone time	Pushed invalidation; short TTL on hot keys
Time-ordered key	The last range, permanently	Writes concentrated on the newest range	Hash prefix, bit-reversed sequence, a different first field

At scale

What	Typical figure
Zipf, `s = 1`, one million keys	Top key 7%, top 100 keys 36%, top 1% of keys 68%
Cluster ceiling with a hot key	Shard capacity ÷ the key’s share: 150,000 ÷ 7% is about 2.1 million ops/s
One relational row under `+ 1` updates	On the order of 1,000 a second
DynamoDB partition	3,000 read units and 1,000 write units a second
Firestore sharded counter	Planned at about 1 write a second per shard document
Local cache, 200 servers, 1 s TTL	200 reads a second per hot key on the shard, whatever the traffic
Fan-out read over N slots, 1% slow each	Slow 8% of the time at N = 8, 47% at N = 64

At 10 times, the hot keys outgrow any one machine. A key at 56,000 a second becomes 560,000, nearly four shards’ worth. So the fixes stack:

Local cache and coalescing for every read.
Replication for the top few keys.
Salting with cached sums for hot counters.
Promotion within 2 seconds.

Known whales get standing treatment, instead of waiting to be detected.

At 100 times, the control loop is a system of its own. It has per-region aggregators, a versioned hot list pushed to every client, a dashboard of what is promoted and why, and an operator override. Sales run on admission, and on pre-minted tokens per region. Some hot keys stop being keys. A celebrity’s recent posts live in memory on every feed server, as in the news feed hybrid that fan-out covers.

What to monitor:

Per-shard load (requests, CPU, network) against the cluster mean. Alert at 1.5 times the mean.
Top keys per shard and their share of it, from the detector.
Time from a shard going hot to its key being promoted, and treatment flips per key.
Shard operations per client request, which rises when salting or replication spread work.
Retries per shard: retries piling onto one shard are the cascade starting.
For sales: admission rate, tokens left per list, and units oversold, which must be zero.

In the interview

Interviewer asksOne key is getting 500,000 reads a second and its shard is melting. Adding shards didn't help. Why not, and what do you do?

A hash sends every request for one key to one shard. So more shards move other keys, and leave this one where it is. The ceiling is one shard’s capacity over the key’s share. I put a local cache with a 1 to 2 second TTL and single flight on every app server. That caps the shard at one read per server per TTL. If the fleet is so large that this still overloads a shard, I replicate the key under 8 suffixes, and read a random one. Writes go to all 8. Public responses go to the CDN. And a detector promotes such keys automatically, because the next one will not be predicted.

Interviewer asksA viral post's like counter is taking 50,000 increments a second. Design the counter.

The like edge stays the truth. Insert (user, post), and increment only if the row is new, so retries do not double count. The count is a sharded counter. The detector promotes the post from 1 slot to 64. Each like increments the slot picked by a hash of its ID, about 800 a second per slot. Reads sum the 64 slots in one batched read, and cache the sum for a second per server. The display rounds to “1.2M”. An hourly job recounts edges for hot posts, to correct drift. If several regions write it, each region gets its own slots, which is a G-counter.

Interviewer asksA flash sale has 10,000 units and 200,000 people pressing buy in the same second. How do you not oversell?

The decrement must stay conditional and exact. So I bound and spread the contention, rather than approximate. A waiting room admits people at the rate the stock rows can take. Admitted buyers pop a token from one of 16 pre-filled lists, one token per unit. Once the lists are empty, everyone else sees “sold out”, with no database work. Stock sits in 16 bucket rows. A purchase decrements a random bucket, with available >= 1 in the WHERE clause, and tries the next if it’s empty. Abandoned tokens return after a timeout. A job reconciles orders, tokens and stock after the sale.

Seen in

Distributed cache: a million reads a second on one key, sampled detection, an L1 near cache, 8-way replication and leases.
URL shortener: a viral link at 50,000 a second, served from in-process caches with single flight.
Instagram: a post at 20,000 likes a second, its counter promoted from one slot to 64.
Ad click aggregator: every ad salted 16 ways by impression ID, with a two-phase merge.
Top-k: views combined at ingest for a second, so a viral video costs about 120 messages a second.
Rate limiter: one tenant’s bucket at up to 300,000 checks a second, held down by token leases and a refusal cache.
Ticketmaster: one event’s seat rows behind a waiting room, and bucketed counters for general admission.
Video streaming: views summed by a stream job into one row per log partition, so a viral video’s writes spread over 100 rows.
Message queue: one partition lagging is almost always one hot key. It’s fixed with a finer key that keeps the order that matters, not with random salting.
News feed: celebrity accounts pulled from memory instead of fanned out.
Leaderboard: one sorted set is one key on one Redis thread, so the board is split 16 ways and merged.
Typeahead: the hottest short prefixes served from the edge.
Key-value store: a key at 200,000 reads a second, cached briefly on coordinators.
Proximity service: the index on every server, so a dense city is never a hot shard.

Recap

The problem. Hashing balances keys, not traffic. A key with share p caps the cluster at shard capacity ÷ p, whatever the shard count. More shards make the hot shard relatively worse.

Sources. Zipf popularity (the top key of a million takes 7%), celebrities and whales, scheduled spikes, viral content, time-ordered keys, and design bugs.

Detect. Sample 1 in 100 on clients, into a bounded top-k. Merge every second, and publish an expiring hot list. Decide on the lower bound, keep a gap between thresholds, and list known hot keys in advance.

Read-hot. A local cache with single flight caps the shard at servers ÷ TTL. Then N copies, and a CDN for public responses. Write-hot. Batch per server. Salt by event hash, with a cached sum. Use CRDT counters across regions, and approximate counts for display. Exact hot rows. Admission, one token per unit, and stock in buckets.

Shed what is left. Failure: the melted shard drags its replica down, stampedes on expiry, salting’s fan-in, late or flapping detection. Stores rebalance ranges, never one key.

Hot keys and hot partitions

The problem it solves

Where skew comes from

Why more shards make it relatively worse

What runs out first

How it works

Step 1: Find the hot keys

Step 2: Read-hot keys: absorb reads before the shard

Step 3: Write-hot keys: spread the writes, or make fewer of them

Step 4: Hot rows that must stay exact: flash sales and inventory

Step 5: Shed what is left

Try it

Options and tradeoffs

Fixes for read-hot keys

Fixes for write-hot keys

Detection

What the store does on its own

How it fails

At scale

In the interview

Seen in

Recap

Further reading

The designbeyondcode digest, every Saturday

Hot keys and hot partitions

The problem it solves

Where skew comes from

Why more shards make it relatively worse

What runs out first

How it works

Step 1: Find the hot keys

Step 2: Read-hot keys: absorb reads before the shard

Step 3: Write-hot keys: spread the writes, or make fewer of them

Step 4: Hot rows that must stay exact: flash sales and inventory

Step 5: Shed what is left

Try it

Options and tradeoffs

Fixes for read-hot keys

Fixes for write-hot keys

Detection

What the store does on its own

How it fails

At scale

In the interview

Seen in

Recap

Further reading

Keep reading

The designbeyondcode digest, every Saturday