system-design · beginner

Hot Partition / Hot Key — When One Shard Takes All the Heat

The Central Question

Consider a sharded social media platform running on the DataLab platform (datalab.com) processing 100,000,000 queries per day:

• The platform shards database rows across 16 database nodes using user_id as the Shard Key.
• Most users receive an average of 5 queries per second.
• Suddenly, a global celebrity (user_id = 9901) posts a breaking announcement. 50,000 users attempt to like, comment on, and read user_id = 9901's post simultaneously.

Meanwhile, the other 15 database shard nodes sit virtually idle at 3% CPU.

This condition is a Hot Key triggering a Hot Partition.

Adding more database shards to the cluster will NOT fix this outage—because all requests for user_id = 9901 will still route exclusively to Shard 4 based on its shard key hash!

This lesson answers one central question: Why do Hot Keys bypass standard database sharding and consistent hashing, and how do engineers eliminate traffic bottlenecks using Key Salting, Local Micro-Caching, and Scatter-Gather Read Aggregation?

Defining the Distinction: Hot Key vs. Hot Partition

While often used interchangeably, engineers distinguish between the root cause (Hot Key) and the system symptom (Hot Partition):

flowchart TD
  Cause[Hot Key: Root Cause] -->|Routes to single node| Symptom[Hot Partition: System Symptom]
  
  Cause --> KeyDesc["A single data item (e.g. Celebrity Post or Viral Product)<br/>receives an extreme spike in read or write QPS."]
  Symptom --> PartDesc["A specific physical database node hits 100% CPU/Disk,<br/>degrading all unrelated keys stored on that node."]

Figure 1: Relationship between root cause Hot Keys and resulting Hot Partitions.

The Partition Skew Ratio ($S_{\text{skew}}$)

$$S_{\text{skew}} = \frac{Q_{\text{max}}}{Q_{\text{avg}}}$$

Where $Q_{\text{max}}$ is the QPS hitting the busiest shard node, and $Q_{\text{avg}}$ is the average QPS across all nodes.

• Healthy Cluster: $S_{\text{skew}} \approx 1.0 - 1.2$ (Traffic is balanced uniformly).
• Hot Partition Incident: $S_{\text{skew}} \ge 10.0$ (One node absorbs 10x more traffic than the average node).

Architectural Mitigation 1: Key Salting (Random Suffixing)

To break a single Hot Key across multiple physical database shards, systems implement Key Salting:

flowchart TD
  subgraph Un-Salted Hot Key Bottleneck
    Client1[50,000 Concurrent Writes] --> Router1{Hash 'post:9901'}
    Router1 -->|100% Traffic| Node4[(Shard Node 4 - 100% CPU CRASH)]
  end
subgraph Salted Key Distribution
    Client2[50,000 Concurrent Writes] --> SaltRouter{Append Random Salt 1..4}
    SaltRouter -->|25% Traffic: post:9901#1| S1[(Shard Node 1)]
    SaltRouter -->|25% Traffic: post:9901#2| S2[(Shard Node 2)]
    SaltRouter -->|25% Traffic: post:9901#3| S3[(Shard Node 3)]
    SaltRouter -->|25% Traffic: post:9901#4| S4[(Shard Node 4)]
  end

Figure 2: Key Salting distributing a viral key across 4 independent database shards.

How Key Salting Works

1. Write Path: When writing a hot counter or post (e.g. post:9901), the application appends a random integer salt between $1$ and $N$ (e.g. post:9901#rand(1,4)).
2. Distribution: Each salted key hashes to a different physical shard server, dividing the 50,000 QPS load into four manageable 12,500 QPS streams across Shard 1, Shard 2, Shard 3, and Shard 4.
3. Read Aggregation: To read the total count, the application queries all $N$ salted keys in parallel (post:9901#1 through post:9901#4) and sums the results.

Architectural Mitigation 2: Local Micro-Caching (L1 Pod RAM)

For read-heavy Hot Keys, querying even a distributed cache cluster (Redis) can cause single Redis node CPU saturation. Systems deploy Local Micro-Caching inside application pod memory:

sequenceDiagram
    autonumber
    actor User as Client Browser
    participant Pod as Application Server Pod
    participant L1Mem as Local RAM (5s Micro-Cache)
    participant DB as Database Shard 4
    
    User->>Pod: 1. GET /posts/9901
    Pod->>L1Mem: 2. Check local sync.Map
    alt L1 Micro-Cache Hit (5-second TTL)
        L1Mem-->>Pod: 3. Return Cached Record (50 microseconds)
        Pod-->>User: 4. HTTP 200 (Instant!)
    else L1 Micro-Cache Miss
        Pod->>DB: 5. Query Database Shard 4 (15ms)
        DB-->>Pod: 6. Return Post Payload
        Pod->>L1Mem: 7. Store in L1 RAM with TTL = 5s
        Pod-->>User: 8. HTTP 200
    end

Figure 3: Local Micro-Caching in application pod RAM absorbing read traffic spikes.

Why Micro-Caching is Extremely Effective

Asynchronous Write-Behind Aggregation for Salted Counters

1. Incoming high-velocity write updates hit random salted keys (counter_key#rand(1,10)) in high-speed Redis RAM.
2. A background worker process periodically runs an aggregation loop every 1 second, executing MGET counter_key#1 ... counter_key#10, summing the totals, and writing the consolidated single total to a primary read-only key counter_key:total.
3. Application clients read counter_key:total in a single $O(1)$ lookup, eliminating scatter-gather fan-out completely.

Single-Tenant Shard Isolation (VIP Tenant Tiering)

• The router detects high-volume enterprise tenant IDs and assigns them to dedicated, isolated single-tenant database shards.
• Small merchants remain on shared multi-tenant shards, insulating them from enterprise traffic surges.

Read Replica Fan-Out Offloading

Detecting Hot Keys via Redis --hotkeys & eBPF Tracing

• Redis LFU --hotkeys Scan: Redis samples key access frequency counters in memory to identify top QPS keys.
• Kernel eBPF Packet Inspection: eBPF probes intercept database network socket packets, calculating per-key query frequency metrics without adding application logging overhead.

Complete Worked Example: Go Hot Key Salted Router & Scatter Aggregator

Let's inspect a complete Go implementation of a Hot Key Salted Router and Scatter Aggregator for the DataLab platform (datalab.com).

package main
import (
	"context"
	"fmt"
	"math/rand"
	"sync"
	"time"
)
type SaltedKeyRouter struct {
	mu         sync.RWMutex
	saltBuckets int
	shardData  map[string]int64 // Simulated Shard Storage: Key -> Counter
}
func NewSaltedKeyRouter(saltBuckets int) *SaltedKeyRouter {
	return &SaltedKeyRouter{
		saltBuckets: saltBuckets,
		shardData:   make(map[string]int64),
	}
}
func (r *SaltedKeyRouter) IncrementSaltedCounter(baseKey string, amount int64) {
	// Append random salt (1..saltBuckets)
	salt := rand.Intn(r.saltBuckets) + 1
	saltedKey := fmt.Sprintf("%s#%d", baseKey, salt)
r.mu.Lock()
	r.shardData[saltedKey] += amount
	r.mu.Unlock()
fmt.Printf("[SALTED WRITE] Distributed increment on %s (+%d)\n", saltedKey, amount)
}
func (r *SaltedKeyRouter) GetAggregatedCounter(baseKey string) int64 {
	var total int64
	var wg sync.WaitGroup
	var mu sync.Mutex
// Parallel Scatter-Gather across all salt buckets
	for i := 1; i <= r.saltBuckets; i++ {
		wg.Add(1)
		go func(bucket int) {
			defer wg.Done()
			saltedKey := fmt.Sprintf("%s#%d", baseKey, bucket)
r.mu.RLock()
			val := r.shardData[saltedKey]
			r.mu.RUnlock()
mu.Lock()
			total += val
			mu.Unlock()
		}(i)
	}
wg.Wait()
	fmt.Printf("[SCATTER AGGREGATE] Aggregated total for %s across %d buckets: %d\n", baseKey, r.saltBuckets, total)
	return total
}

Failure Modes and Engineering Mitigations

What You Should Remember

1. Hot Keys bypass standard sharding: Adding more database shards does not fix a Hot Key because all requests route to the same node based on shard key hash.
2. Quantify traffic skew with Partition Skew Ratio: Monitor $S_{\text{skew}} = \frac{Q_{\text{max}}}{Q_{\text{avg}}}$ to detect hot partitions before node crashes occur.
3. Use Key Salting for hot writes: Append random suffixes (key#rand(1,N)) to scatter hot writes across $N$ independent physical shards.
4. Deploy Local Micro-Caching for hot reads: Cache hot records in application pod RAM for 1-5 seconds to collapse 50,000 QPS down to 20 QPS on databases.
5. Isolate Enterprise Hot Tenants: Move massive single-tenant customers off shared multi-tenant shards onto dedicated single-tenant database nodes.

Glossary of Terms

Practice Scenario and Self-Assessment

Architecture Scenario

• 500,000 fans attempt to purchase tickets for a popular concert (event_id = 4092) at 10:00 AM.
• The inventory counter event_4092_seats receives 50,000 write updates per second.

1. Formulate the Key Salting and Scatter Aggregation architecture to process 50,000 seat reservation updates per second.
2. Detail how your API prevents overselling when combining Key Salting with asynchronous bucket aggregation.

Interactive Self-Assessment

All 50,000 QPS for that key still map to the exact same shard node based on its shard key hash.

Adding shards automatically formats the underlying NVMe SSD disk drives.

Adding shards revokes edge HTTPS TLS encryption certificates on load balancers.

Adding shards cuts physical CPU hardware clock speeds in half across all nodes.

Each application pod serves reads from local RAM, querying the database at most once every 3 seconds per pod.

Local micro-caching automatically converts relational SQL schemas into un-indexed CSV files.

Local micro-caching replaces public DNS nameservers with local hosts entries.

Local micro-caching reboots operating system hypervisors across all application pods.

What to Learn Next

• Rebalancing Shards Under Skewed Traffic: Master live data migration and range splitting.
• Database Sharding — Split Data Across Many Machines: Review shard keys and scatter-gather queries.
• Database Scaling — The Progressive Scaling Ladder: Revisit query optimization and read-replica splitting.

Track: Data, Storage and Messaging

Previous: Database Sharding — Split Data Across Many Machines

Next: How to Scale a Database — The Progressive Scaling Ladder

By Shubham Jain

All articles · Study paths

Shubham Jain · Learning Lab