You did everything right.
Split the database into 16 shards. Distributed users evenly by user_id hash. Each shard handles 6.25% of traffic. Perfect balance.
Then Black Friday happened.
One celebrity with 50 million followers posted about your product. All 50 million followers have user IDs that hash to... shard 7.
Shard 7 is now handling 80% of your traffic. The other 15 shards are idle. Shard 7 is melting.
Welcome to the Hot Partition Problem.
Why Hashing Isn't Enough
Hash-based sharding looks perfect on paper:
def get_shard(user_id):
return hash(user_id) % num_shards
Uniform distribution. Simple logic. What could go wrong?
Everything. Because real-world access patterns don't care about your hash function.
Scenario 1: Celebrity Effect
A viral post from one user means millions of reads on that user's shard. Followers are distributed across shards, but the content they're accessing isn't.
Scenario 2: Time-Based Clustering
Users who signed up on the same day often have sequential IDs. They also often have similar usage patterns. Your "random" distribution isn't random at all.
Scenario 3: Geographic Hotspots
Morning in Tokyo means heavy traffic from Japanese users. If your sharding key correlates with geography, one shard gets hammered while others sleep.
How to Detect Hot Partitions
You can't fix what you can't see.
Monitor per-shard metrics:
Shard 1: CPU 15% | QPS 1,200 | Latency P99 45ms
Shard 2: CPU 12% | QPS 1,100 | Latency P99 42ms
Shard 7: CPU 94% | QPS 18,500 | Latency P99 890ms ← PROBLEM
Shard 8: CPU 18% | QPS 1,400 | Latency P99 51ms
Set up alerts:
- Single shard CPU > 70% while others < 30%
- Single shard latency > 3x average
- Single shard QPS > 5x average
Track hot keys:
Log the most frequently accessed keys per shard. The top 1% of keys often cause 50% of load.
Solution 1: Add Randomness to Hot Keys
For keys you know will be hot, add a random suffix:
def get_shard_for_post(post_id, is_viral=False):
if is_viral:
# Spread across multiple shards
random_suffix = random.randint(0, 9)
return hash(f"{post_id}:{random_suffix}") % num_shards
else:
return hash(post_id) % num_shards
A viral post now spreads across 10 shards instead of 1. Reads are distributed. Writes need to fan out, but that's usually acceptable.
The tricky part: knowing which keys will be hot before they're hot.
Solution 2: Dedicated Hot Shard
Accept that some data is special. Give it special treatment.
HOT_USERS = {"celebrity_1", "celebrity_2", "viral_brand"}
def get_shard(user_id):
if user_id in HOT_USERS:
return HOT_SHARD_CLUSTER # Separate, beefier infrastructure
return hash(user_id) % num_shards
The hot shard cluster has more replicas, more CPU, more memory. It's designed to handle disproportionate load.
Update the HOT_USERS list dynamically based on follower count or recent engagement metrics.
Solution 3: Caching Layer
Don't let hot reads hit the database at all.
def get_post(post_id):
# Check cache first
cached = redis.get(f"post:{post_id}")
if cached:
return cached
# Cache miss - hit database
post = database.query(post_id)
# Cache with TTL based on hotness
ttl = 60 if is_hot(post_id) else 300
redis.setex(f"post:{post_id}", ttl, post)
return post
For viral content, a 60-second cache means the database sees 1 query per minute instead of 10,000 queries per second.
Shorter TTL for hot content sounds counterintuitive, but it ensures fresher data for content people actually care about.
Solution 4: Read Replicas Per Shard
Scale reads horizontally within each shard:
Shard 7 Primary (writes)
├── Replica 7a (reads)
├── Replica 7b (reads)
├── Replica 7c (reads)
└── Replica 7d (reads)
When shard 7 gets hot, spin up more read replicas for that specific shard. Other shards stay lean.
This works well for read-heavy hotspots. Write-heavy hotspots need different solutions.
Solution 5: Composite Sharding Keys
Don't shard on a single dimension:
# Bad: Single key sharding
shard = hash(user_id) % num_shards
# Better: Composite key
shard = hash(f"{user_id}:{content_type}:{date}") % num_shards
Composite keys add entropy. A celebrity's posts are now spread across shards by date, not concentrated in one place.
The trade-off: queries that span multiple values need to hit multiple shards. Design your access patterns accordingly.
Solution 6: Dynamic Rebalancing
When a partition gets hot, split it:
Before:
Shard 7 handles hash range [0.4375, 0.5000]
After split:
Shard 7a handles [0.4375, 0.4688]
Shard 7b handles [0.4688, 0.5000]
Modern distributed databases like CockroachDB and TiDB do this automatically. If you're running your own sharding, you'll need to build this logic.
Key considerations:
- Data migration during split
- Connection draining
- Query routing updates
Prevention Checklist
Before your next traffic spike:
1. Know your hot keys
Run analytics on access patterns. Which users, which content, which time periods drive disproportionate load?
2. Design for celebrities
If your product could have viral users, plan for them. Don't wait until you have one.
3. Monitor per-shard, not just aggregate
Average latency across 16 shards hides the shard that's dying. Track each one individually.
4. Test with realistic skew
Load tests with uniform distribution prove nothing. Simulate 80% of traffic hitting 5% of keys.
5. Have a manual override
When detection fails, you need a way to manually mark keys as hot and reroute them.
The Reality
Perfect distribution doesn't exist in production.
Users don't behave uniformly. Content doesn't go viral uniformly. Time zones don't align uniformly.
Your sharding strategy needs to handle the 99th percentile, not the average. One hot partition can take down your entire system while 15 other shards sit idle.
Design for imbalance. Monitor for hotspots. Have a plan before the celebrity tweets.
Further Reading
For comprehensive patterns on building resilient distributed databases—including sharding strategies, replication topologies, and connection management for high-traffic platforms:
→ Enterprise Distributed Systems Architecture Guide
16 shards. Perfect hashing. One celebrity. One fire.
Top comments (0)