Scaling Sharded Databases: Taming Hot Shards with Dynamic Rebalancing

A hot shard is a database partition that receives disproportionately high load due to skewed traffic patterns. In ad tech, this occurs when a few large advertisers or campaigns dominate traffic. A single popular advertiser might generate 10% of all impressions; if the advertiser is sharded by a simple hash(advertiser_id), that advertiser's shard must handle 10x the average load. This shard becomes a bottleneck: query latency spikes, disk I/O maxes out, and the entire system is rate-limited by the slowest shard (Amdahl's law).

The severity scales with platform size. At 1 billion impressions per day, 10% concentration means 100 million impressions for one shard while others handle 10 million. If a shard can handle 50k QPS, the hot shard hits saturation at 50k QPS while other shards are idle. This wastes expensive infrastructure and degrades advertiser experience.

The challenge is that the skew is not uniform: the top 1% of advertisers generate 30–50% of impressions; the distribution is Zipfian. Worse, skew changes over time (a new product launch can suddenly become dominant, or a major advertiser can scale down). Any static partitioning scheme will fail.

Three approaches exist:

(1) reduce load per advertiser via caching and aggregation,

(2) distribute one advertiser across multiple physical shards (virtual sharding),

(3) separate hot advertisers into dedicated infrastructure. Each has different cost/complexity trade-offs.

Key Requirements

Functional Requirements

• Consistent routing: same advertiser always routes to same shard (for transactional consistency)
• Hot shard detection: identify skewed advertisers/campaigns in real-time
• Load shedding: gracefully degrade when shard is overloaded (queue requests, not drop)
• Rebalancing: move data between shards without downtime
• Cross-shard consistency: if advertiser spans multiple shards, maintain consistency of budget/spend tracking
• Per-tenant isolation: one advertiser's traffic spike should not affect others

Non-Functional Requirements

• Latency SLA: maintain <100ms p99 latency even with 10x traffic spike on hot shard
• Throughput: support 1B+ impressions/day distributed across 32 shards; individual shard capacity 50k QPS
• Load imbalance tolerance: top 1% of advertisers (by traffic) should not cause shard utilization >80%
• Detection latency: identify hot shard within 5 minutes of traffic spike
• Rebalancing time: migrate advertiser to new shard in <30 minutes with zero downtime
• Cost: avoid over-provisioning idle shards; aim for average 50–60% utilization

Simplified Architecture

Architecture Proposals

Option 1: Hash Salting with Virtual Sharding

Instead of hashing directly to physical shards (hash(advertiser_id) % num_shards), use virtual sharding: hash to a larger virtual shard space (e.g., 10,000 virtual shards), then map virtual shards to physical shards with flexible mapping.

The flow is:

1. Request arrives for advertiser_id: virtual_shard = hash(advertiser_id) % 10000
2. Lookup mapping table (in DynamoDB or ElastiCache): virtual_shard → physical_shard (e.g., physical_shard = virtual_shard % 32)
3. Query physical shard
4. On hot shard detection: remapping. Instead of mapping 10000 → 32, split hot virtual shards: virtual_shards 2000–2500 map to new physical shard 33. Load on previous shard decreases.

The key insight: by introducing a layer of indirection (virtual shards), you can rebalance without rehashing the entire dataset.

Pros

• Fine-grained load balancing: can split a single advertiser across multiple shards by remapping virtual shards
• No rehashing required: virtual shards are stable; only mapping table changes
• Incremental rebalancing: can move 1% of load at a time without data migration
• Low latency overhead: mapping lookup is sub-1ms (DynamoDB or in-memory cache)
• Backward compatible: existing hash function doesn't change; only routing layer adapts

Cons

• Requires mapping table management: must maintain and update mapping in real-time; potential consistency issues if mapping is stale
• Virtual shard count tuning: too few (e.g., 100) and splitting is coarse; too many (e.g., 100,000) and mapping table is huge
• Detection complexity: monitoring must identify hot physical shards and recommend remapping; automated detection is error-prone
• Data locality lost: if an advertiser's data was colocated on one shard, splitting it across shards breaks locality. Query performance degrades.
• Cache invalidation: routing layer caches mapping table; stale caches cause requests to old shard. Needs cache TTL and invalidation mechanism.

Failure Modes & Mitigations

• Mapping table out of sync: client has cached mapping, but mapping changed. Request routes to old shard which no longer holds the advertiser. Risk: data loss or 404. Mitigation: mapping change is a two-phase process. (1) New shard accepts requests for advertiser, (2) old shard rejects with "moved to X" error, client retries. Keep old shard for 24h grace period.
• Split is too aggressive: remap 50% of virtual shards to new shard. New shard is overwhelmed. Risk: cascading failure. Mitigation: limit rebalancing to 10% of load per operation; monitor new shard latency post-remapping via CloudWatch.
• Virtual shard count mismatch: 10000 virtual shards, but only 32 physical shards. Some physical shards map to many virtual shards (high load), others to few (idle). Risk: uneven distribution. Mitigation: ensure virtual shard count is multiple of physical shard count (e.g., 10000 / 32 = 312.5; use 10240 = 32 * 320). Use consistent hashing for better distribution.

When to Choose Choose this if you have existing Aurora shard infrastructure and want to add dynamic rebalancing without major refactoring. Good for 10–100 hot advertisers (top 1% by traffic). Requires someone to write mapping table management logic and monitoring. Avoid if your data model requires strong cross-shard transactions (ACID across shards is hard with split data).

Option 2: Read-Through Cache Layer (ElastiCache Redis / DynamoDB DAX)

Instead of querying the database shard directly, route all reads through a cache layer. The cache is much faster than the database; it absorbs read-heavy traffic (statistics, account balance queries), reducing load on the hot shard.

The flow is:

1. Request arrives for advertiser_id (e.g., "what is my spend so far today?")
2. Cache lookup via ElastiCache: GET advertiser_id:spend. If hit, return immediately (<1ms).
3. If miss: query Aurora database shard. Response: spend = $1234.56. SET advertiser_id:spend = "$1234.56" in ElastiCache with EX 300 (5 min TTL).
4. Return response.

For write-heavy operations (ad delivery decisions), cache is less effective; but read queries (dashboard, reporting, account management) are often 60–80% of traffic.

Pros

• Simple to implement: add ElastiCache in front of database; applications don't change much
• Reduces database load: cache absorbs 60–80% of reads; hot shard receives 20–40% of original load
• Fast response time: cache hits are <1ms; misses are slow (100ms), but overall average improves
• Operational simplicity: ElastiCache is well-understood; no complex remapping logic
• Per-advertiser granularity: different advertisers can have different cache policies (hot advertiser: longer TTL; cold: no cache)
• Works well with eventual consistency: cache is eventually consistent (data may be stale 5 mins)

Cons

• Write-through cost: writes must hit database; cache is invalidated. Write-heavy workloads don't benefit.
• Stale data: if advertiser's spend updates but cache isn't invalidated, queries return old balance. Can break billing logic if not careful.
• Cache invalidation complexity: need to invalidate cache when advertiser's data changes. Pattern: cache-aside (lazy invalidation with TTL) vs. write-through (eager invalidation on writes) both have issues.
• Hot cache itself: if advertiser is very popular, all reads hit the same ElastiCache shard (redis key = advertiser_id). ElastiCache shard becomes hot! (recursive problem). Solution: shard ElastiCache differently (by region, not by advertiser).
• Memory cost: cache stores full dataset; expensive for large (advertiser_id → balance) mappings. Requires aggressive TTL and LRU eviction.

Failure Modes & Mitigations

• Cache stampede: popular advertiser's cache key expires; 1000 requests arrive simultaneously, all miss cache, all query database. Risk: database spike (thundering herd). Mitigation: use probabilistic early expiration (refresh cache before TTL expires); or use request coalescing (first request populates cache, others wait for result).
• Stale data served: advertiser's spend updated, but cache not invalidated. Dashboard shows old balance. Risk: advertiser confusion, billing disputes. Mitigation: use short TTL (5 min max); mark cache as stale if source data was updated recently; or use write-through pattern (always update cache on writes).
• Cache memory explosion: 10M advertisers × 1KB per advertiser = 10 GB. Can't fit in ElastiCache. Risk: eviction thrashing. Mitigation: cache only hot advertisers (top 10k by QPS); cold advertisers skip cache; use Bloom filter to avoid cache lookup for non-existent keys.

When to Choose Choose this for read-heavy platforms (reporting, analytics, account queries). Ideal for existing Aurora databases that can't be easily sharded or rebalanced. Works well if hot advertisers have predictable, repeating queries (same balance lookup). Avoid if workload is write-heavy or if stale data is unacceptable (billing, audit logs).

Option 3: Per-Tenant Sub-Clusters (Dedicated Aurora Clusters for Largest Advertisers)

The most extreme approach: separate the largest advertisers into dedicated, isolated infrastructure. Top 100 advertisers each get their own Aurora cluster (or at least dedicated shards on a separate Aurora cluster). Remaining advertisers share a multi-tenant cluster.

The flow is:

1. Request for advertiser_id: lookup advertiser tier (top-100, regular)
2. If top-100: route to dedicated cluster (e.g., advertiser_cluster_42 in Route 53)
3. If regular: route to shared cluster, hash to shard
4. Execution: each cluster is independent, scales independently

This provides hard isolation: one advertiser's spike doesn't affect others.

Pros

• Hard isolation: one advertiser's issues don't affect others
• Independent scaling: dedicated cluster can be tuned for advertiser's workload (e.g., top advertiser uses faster CPUs)
• Predictable SLA: each advertiser has guaranteed capacity; no oversubscription
• Operational control: can run different versions, backups, tuning per advertiser
• Revenue protection: ensure largest customers get best infrastructure

Cons

• High cost: 100 dedicated Aurora clusters means 100x licensing, backup infrastructure, monitoring. Estimate $5–10k per month per cluster = $600k–$1.2M per year.
• Operational burden: 100 independent clusters to manage, patch, monitor. On-call rotation becomes massive.
• Utilization waste: small dedicated clusters often underutilized (30–40% CPU). Money wasted on idle capacity.
• Data replication complexity: if advertiser data needs to be accessed from multiple regions, replicate across 100 clusters becomes complex.
• Migration risk: moving advertiser from shared to dedicated cluster requires data migration, potential downtime.
• Fairness perception: "why does advertiser X get dedicated cluster but I don't?" Can lead to feature requests and customer support burden.

Failure Modes & Mitigations

• Dedicated cluster fails: advertiser cluster goes down. Risk: complete service outage for top advertiser. Mitigation: use Aurora Global Database (cross-region replicas); active-active replication with conflict resolution; cross-region failover (2+ regions per advertiser).
• Capacity planning mismatch: dedicated cluster is provisioned for Q2 traffic, but Q4 surge exceeds capacity. Risk: underprovisioned. Mitigation: provision for 2x peak (expensive); or use Aurora Auto Scaling with manual overrides for burstiness.
• Stranded tenants: advertiser moves to shared cluster after contract ends, but nobody cleans up dedicated cluster. Risk: idle clusters burn money. Mitigation: automation: when advertiser tier drops below top-100, migrate data to shared cluster; delete dedicated cluster via Lambda.

When to Choose Choose this only if your largest advertiser generates >20% of revenue and needs guaranteed SLAs. Ideal for high-touch enterprise accounts. Works well if you have a dedicated SRE team for enterprise customers. Avoid unless you are confident the cost and operational burden are justified by revenue.

Comparison Table

Recommendation

For most platforms, start with Option 2 (Cache Layer), then migrate to Option 1 (Hash Salting) as complexity grows.

Option 2 is best if:

• Your workload is 60%+ reads (dashboard, reporting, account queries)
• Top 10 advertisers generate <30% of traffic (no single dominating advertiser)
• You need a quick fix (<2 weeks implementation)

Upgrade to Option 1 (Hash Salting) when:

• You have 100M+ impressions/day and top advertisers are creating hot shards
• Your team can implement mapping table management and rebalancing logic
• You need fine-grained, dynamic rebalancing without data migration

Use Option 3 (Dedicated Clusters) only if:

• Your largest advertiser generates >25% of revenue
• That advertiser has an SLA contract requiring guaranteed capacity
• You have a dedicated enterprise ops team
• The $5M+/year cost is justified by revenue

Implementation advice for Option 1 (recommended):

1. Use 10,240 virtual shards (32 * 320), map to 32 physical Aurora shards initially
2. Store mapping in DynamoDB: virtual_shard → physical_shard (or in-memory cache with TTL)
3. Cache mapping in application with 10-minute TTL; monitor cache hit rate via CloudWatch
4. Monitor physical shard metrics (QPS, latency, disk I/O) every minute via CloudWatch dashboards
5. If shard exceeds 70% utilization, alert on-call; if >80%, auto-remap 10% of hot virtual shards to new shard (or ask on-call first)
6. Implement two-phase remapping: write to new shard with "provisional" flag, read from both shards for validation, commit mapping change, grace period 24h on old shard
7. Log all remaps; review weekly for trends (which advertisers are hot, when, why)

Capacity & Performance Notes

Sizing for 1B impressions/day with 100k advertisers (Option 1):

• Shard distribution: Zipfian distribution with alpha=1.5. Top 1% (1000 advertisers) generate 35% of traffic = 350M impressions. Average: 350M / 1000 = 350k impressions per advertiser.
• Virtual shard count: 10,240 (for 32 physical shards)
• Per physical shard: 10,240 / 32 = 320 virtual shards. If evenly distributed, each shard handles ~31M impressions/day ≈ 359 QPS average. Peak (2x) = 718 QPS, acceptable for r6g.4xlarge (50k QPS capacity).
• Mapping table: 10,240 entries × 16 bytes (virtual_shard_id + physical_shard_id) = 160 KB. Fits in L1 cache; negligible lookup latency.
• Rebalancing example: top advertiser is hot (5% of traffic = 50M impressions = 579 QPS). Remap 100 virtual shards (1/100 of load) from shard 5 to shard 33. Each shard goes from 359 to ~340 and ~378 QPS respectively. Acceptable.

Option 2 sizing (Cache Layer):

• Read ratio: assume 70% of requests are reads (stats, balance, account settings); 30% are writes (bid delivery, budget updates)
• Cache hit ratio: 95% for reads (same advertiser queries repeatedly)
• Load reduction: 70% × 95% = 66.5% of requests avoid database. Hot shard load drops 66.5%, from 1000 QPS to 335 QPS.
• Cache size: 100k advertisers × 1 KB (balance, stats, settings) = 100 GB. Too large. Solution: only cache top 10k advertisers. 10k × 1 KB = 10 GB. Use ElastiCache Cluster Mode with 4 shards, each 4 GB, on cache.r6g.xlarge.
• TTL: 5 minutes for balance (acceptable staleness); 24 hours for rarely-changing settings.

Option 3 sizing (Dedicated Clusters):

• Per-cluster: top advertiser generates 50M impressions/day = 579 QPS. Provision 2x = 1200 QPS capacity. Single r6g.4xlarge (50k QPS) is enough, but use 2 for HA. Plus 2 read replicas. 4 instances × $0.50/hour = $14.4k per month per advertiser.
• For 100 advertisers: $1.44M per month = $17.3M per year. Add 10% overhead (shared infrastructure, tooling) = $19M/year.