Bird
Raised Fist0

Consistent Hashing for Database Sharding at Scale

HLDmediumInterview freq: common
Choose your preparation mode4 modes available
ArchitectureTraceChallenge
🏗️
Consistent Hashing for Database Sharding at Scale
Design a scalable, fault-tolerant consistent hashing scheme for database sharding to handle billions of users and high QPS
mediumConsistent HashingFreq: common
Prerequisites:Distributed Systems BasicsHashingDatabase ShardingCAP Theorem
⚙️
Functional Requirements
  • Distribute data across multiple database nodes using consistent hashing to minimize data movement on node changes
  • Support automatic re-sharding when nodes are added or removed with minimal impact on availability
  • Provide a lookup service to map keys to shard nodes with low latency (<5ms)
  • Ensure data consistency and availability during node failures and rebalancing
  • Support replication of data across multiple nodes for fault tolerance
  • Allow dynamic scaling of shards to handle traffic growth without downtime
📊
Non-Functional Requirements
  • Handle 2 billion daily active users generating 40 writes per user per day
  • Achieve 99.99% availability with failover and minimal downtime during re-sharding
  • Maintain 100ms P99 latency for read and write requests
  • Support peak write throughput of approximately 925K writes per second
  • Ensure data rebalancing completes within 10 minutes when scaling nodes
❓
Clarifying Questions to Ask
  • What is the expected data size per user or per key on average?
  • Are we supporting strong consistency or eventual consistency for reads/writes?
  • What is the replication factor for fault tolerance?
  • How frequently do we expect nodes to be added or removed?
  • Is the system read-heavy, write-heavy, or balanced?
  • What is the acceptable downtime or performance degradation during re-sharding?
Out of Scope
✗ Designing the underlying database storage engine (e.g., Cassandra internals)
✗ Implementing application-level caching or CDN layers
✗ Designing client SDKs or APIs beyond shard lookup
✗ Security and encryption of data at rest or in transit
📌
Assumptions
  • 2 billion daily active users (DAU)
  • Each user generates 40 writes per day
  • Average message size is 100 bytes
  • Replication factor is 3 for fault tolerance
⚡
QPS Calculation
write qps2B × 40 / 86400 ≈ 925,926 writes/sec
peak write qpsAssuming 2× peak factor: 925,926 × 2 ≈ 1,851,852 writes/sec
read qpsAssuming 3 reads per write: 925,926 × 3 ≈ 2,777,778 reads/sec
💾
Storage Estimate
daily storage2B × 40 × 100 bytes = 8,000 TB/day ≈ 7.45 PB/day
retention days7
total storage7.45 PB × 7 = 52.15 PB
with replication52.15 PB × 3 = 156.45 PB total replicated storage
🖥️
Server Estimate
shard nodesAssuming each node handles 10K writes/sec, need 185 nodes for peak writes
lookup serviceUsing 10 lookup servers with <5ms latency to handle 3M reads/sec with load balancing
🏗️
Architecture Diagram
Architecture Diagram
          +----------------------+
          |   Client Applications |
          +----------+-----------+
                     |
                     v
          +----------------------+
          | Consistent Hash Ring |
          |  (Hashing + Lookup)  |
          +----------+-----------+
                     |
         +-----------+------------+
         |                        |
 +---------------+        +---------------+
 |  Shard Node 1 |        |  Shard Node N |
 | (Cassandra)   |        | (Cassandra)   |
 +-------+-------+        +-------+-------+
         |                        |
         +-----------+------------+
                     |
          +----------------------+
          | Replication Manager  |
          +----------------------+
                     |
          +----------------------+
          | Monitoring & Metrics |
          +----------------------+
🧩
Components
Consistent Hash Ring Service
Maps keys to shard nodes using consistent hashing with virtual nodes
Why thisConsistent hashing minimizes data movement on node changes and balances load using virtual nodes
Why notRange-based sharding requires expensive rebalancing and causes hotspots under skewed keys
Shard Nodes (Apache Cassandra)
Store and serve data partitions assigned by consistent hashing
Why thisCassandra supports horizontal scaling, tunable consistency, and high write throughput needed for 1.8M writes/sec
Why notPostgreSQL requires manual sharding and does not scale linearly at this write QPS
Replication Manager
Ensures data replication across multiple shard nodes for fault tolerance and consistency
Why thisCustom replication logic integrated with Cassandra's replication to handle node failures and re-sharding
Why notRelying solely on Cassandra's internal replication lacks control over re-sharding impact and failover timing
Lookup Service (Redis Cluster)
Provides low-latency key-to-shard mapping for clients to route requests
Why thisRedis Cluster supports fast in-memory lookups with TTL and automatic failover for high availability
Why notUsing a database for lookup adds latency and complexity; DNS-based lookup is too slow and inconsistent
Monitoring & Metrics (Prometheus + Grafana)
Tracks system health, latency, throughput, and triggers alerts on anomalies
Why thisPrometheus provides scalable metrics collection with Grafana dashboards for real-time observability
Why notCustom monitoring lacks ecosystem support and scalability at 2B DAU scale
🔄
Data Flow
1
Client Application → Lookup Service
Client queries lookup service with key to get shard node address
Lookup service uses consistent hash ring to return shard responsible for key
2
Client Application → Shard Node
Client sends read/write request directly to shard node
Direct communication reduces latency and load on lookup service
3
Shard Node → Replication Manager
Shard node replicates data to other nodes as per replication factor
Replication ensures fault tolerance and data durability
4
Replication Manager → Shard Nodes
Replication manager coordinates data sync and handles failover
Maintains consistency and availability during node failures
5
Monitoring & Metrics → Operations Team
Collects metrics and alerts on latency, throughput, and failures
Enables proactive detection and response to system issues

Practice

(1/5)
1. In a Count-Min Sketch system, at step 3 the Update Worker hashes an incoming event key with the second hash function. What happens to the corresponding counter in the sketch at this step?
easy
A. The counter is reset to zero before incrementing
B. The counter at the hashed index is incremented by one
C. The counter is decremented to avoid overcounting
D. The counter is replaced with the maximum of current and new value

Solution

  1. Step 1: Understand Update Worker hashing

    Each hash function maps the event key to an index in the sketch array.

  2. Step 2: Increment counter at hashed index

    The Count-Min Sketch increments the counter at that index to record the event occurrence.

  3. Final Answer:

    Option B -> Option B

  4. Quick Check:

    Update increments counters at hash indices to track frequencies [check]
Hint: Update increments counters at hash indices, never resets or decrements [check]
Common Mistakes:
2. In the adaptive streaming request flow, at step 4 the CDN edge server checks the client's current bandwidth estimate. What does the CDN do if the bandwidth is below the threshold for HD quality?
easy
A. It immediately switches the stream to the lowest bitrate available without buffering
B. It pauses the stream and waits for the client to buffer more data
C. It serves the next video chunk at a lower bitrate from its cache if available
D. It requests the origin server to send a new manifest with lower bitrate options

Solution

  1. Step 1: Understand CDN adaptive streaming role at step 4

    The CDN edge server uses bandwidth info to decide streaming quality dynamically.

  2. Step 2: Analyze adaptive streaming behavior on low bandwidth

    Switching immediately to lowest bitrate (It immediately switches the stream to the lowest bitrate available without buffering) can cause playback stalls. Requesting new manifest (It requests the origin server to send a new manifest with lower bitrate options) is unnecessary at CDN edge. Serving lower bitrate chunk from cache (It serves the next video chunk at a lower bitrate from its cache if available) requires prior caching. The CDN typically pauses streaming to allow client buffering (It pauses the stream and waits for the client to buffer more data) to avoid playback interruption.

  3. Final Answer:

    Option B -> Option B

  4. Quick Check:

    CDN pauses streaming to buffer on low bandwidth [check]
Hint: Pause to buffer before switching bitrate [check]
Common Mistakes:
3. If the authentication service in a client-server system becomes unavailable, what is the exact failure mode experienced by clients attempting to log in?
medium
A. Clients receive a generic server error but can retry immediately without data loss
B. Clients are authenticated locally by the server, so login proceeds without interruption
C. Clients' login requests hang until timeout, then receive authentication failure with no retry
D. Clients are redirected to a backup authentication service automatically

Solution

  1. Step 1: Understand authentication dependency

    The server relies on the authentication service for validating credentials; if unavailable, requests cannot be processed.

  2. Step 2: Trace failure impact

    Without authentication service, login requests hang waiting for response until timeout, then fail with no automatic retry or fallback.

  3. Final Answer:

    Option C -> Option C

  4. Quick Check:

    Unavailable auth service causes login hang then failure [check]
Hint: No auth service -> login hangs then fails, no fallback [check]
Common Mistakes:
4. Without the retry queue component in the payment gateway's webhook delivery system, what exact failure occurs when a merchant's endpoint is temporarily down?
medium
A. Webhook notifications are retried immediately in a tight loop, overwhelming the merchant server
B. Webhook delivery blocks the main payment processing thread until the merchant recovers
C. Webhook notifications are queued in memory indefinitely, causing memory leaks and crashes
D. Webhook notifications are lost permanently with no retry, causing inconsistent merchant state

Solution

  1. Step 1: Understand retry queue role

    The retry queue buffers failed webhook notifications for asynchronous retries, ensuring eventual delivery.

  2. Step 2: Consequence of missing retry queue

    Without it, failed webhooks are dropped or lost, causing permanent notification loss and inconsistent merchant state.

  3. Final Answer:

    Option D -> Option D

  4. Quick Check:

    Missing retry queue -> permanent webhook loss on failure [check]
Hint: Retry queue prevents permanent webhook loss [check]
Common Mistakes:
5. At 10x scale, which component in a consistent hashing system with virtual nodes is most likely to fail first and why?
hard
A. The physical nodes, because they cannot handle increased key storage and request load
B. The token metadata service, due to increased complexity managing many virtual nodes and frequent updates
C. The network, because consistent hashing increases network chatter exponentially with scale
D. The client-side hash function, as hashing large key sets becomes computationally infeasible

Solution

  1. Step 1: Identify scaling bottlenecks

    Physical nodes scale horizontally; network and hash function scale well with optimizations.

  2. Step 2: Analyze token metadata service

    Managing many virtual nodes and frequent node changes increases metadata complexity and update frequency, causing bottlenecks.

  3. Final Answer:

    Option B -> Option B

  4. Quick Check:

    Token metadata service complexity grows fastest, causing first failure [check]
Hint: Token metadata complexity grows fastest at scale [check]
Common Mistakes:
Recommended

Ready to prove it?

Pass the test and this pattern moves unknown strong while earning +1 Grip toward your certificate.

10 questions~5 mineasy hard