Splitting data across nodes to improve
scalability (horizontal growth),
throughput (parallel processing), and
storage capacity (distribute data),
with trade-offs in
complexity (routing, rebalancing) and
skew (uneven distribution).
▸ Horizontal vs Vertical Partitioning
Horizontal (Split Rows)
Vertical (Split Columns)
▸ Partitioning Strategies
Range
Hash
List (Explicit)
Guarantees: Partitioning provides linear write scaling. Partition pruning skips irrelevant partitions. Cross-partition queries are expensive (scatter-gather).
Hotspot Mitigation: Celebrity problem — fix: add random suffix → scatter across shards → aggregate on read.
Distributed Indexing
Indexing data across nodes to improve
query performance (faster lookups),
scalability (handle large datasets), and
throughput (parallel queries),
with trade-offs in
consistency (index lag),
write overhead (maintain indexes), and
complexity (coordination).
How indexes work across partitions and shards — local vs global indexes
In a single DB, an index covers all rows. Once you partition/shard, the index must either live inside each partition (local) or span across all partitions (global). This choice fundamentally affects query routing, write latency, and consistency.
▸ Local Index vs Global Index
Local Index (per-partition)
Pro Fast writes — index updated locally, no cross-partition coordination Con Non-partition-key queries → scatter-gather (query ALL partitions) Used by: Cassandra, DynamoDB, MongoDB, Postgres partitions
Global Index (cross-partition)
Pro Fast reads — index tells you exactly which partitions to hit Con Slower writes — every write must update the global index (cross-partition) Used by: DynamoDB GSI, Google Spanner, CockroachDB, Citus
Aspect
Local Index
Global Index
Write speed
Fast — update local index only
Slower — must update global index (cross-partition)
Read (by partition key)
Fast — single partition
Fast — single partition
Read (non-partition key)
Scatter-gather — query ALL partitions
Targeted — index routes to exact partitions
Consistency
Strong (same partition)
Eventual (async index update) or strong (sync, slower)
Examples
Cassandra, MongoDB, Postgres partitions
DynamoDB GSI, Spanner, CockroachDB
▸ How Indexing Fits with Sharding & Replication
Rule of thumb: Query by partition key → local index is enough (fast). Query by non-partition column → you need a global secondary index (DynamoDB GSI) or accept scatter-gather. Search workloads → use a dedicated search engine (Elasticsearch) with its own distributed inverted index.
Replication
Copying data across nodes to improve
availability (failover),
fault tolerance (survive failures),
durability (prevent data loss),
disaster recovery (region backup), and
read scalability (higher read throughput),
with trade-offs in
consistency (stale reads),
write latency (slower writes), and
operational complexity (manage replicas).
▸ Sync vs Async vs Semi-Sync Replication
Synchronous
Ex: Spanner (TrueTime), Postgres synchronous_commit=on. Banking, payments — can't lose a single write.
Asynchronous
Ex: MySQL default, Postgres async replicas, MongoDB secondaries. Social media feeds, analytics — speed over durability.
Semi-Synchronous
Ex: MySQL semi-sync, Postgres synchronous_standby_names (1 of N). E-commerce orders — need durability but not full sync overhead.
Mode
Client Waits For
Data Loss Risk
Write Latency
Use Case
Sync
ALL replicas ACK
Zero
Highest
Banking, payments (Spanner)
Semi-Sync
1 replica ACK
Near-zero
Medium
E-commerce, orders (MySQL semi-sync)
Async
Master only
Possible (lag window)
Lowest
Social feeds, analytics (Postgres async)
▸ Single-Leader Replication — Deep Dive
▸ Multi-Leader Replication — Deep Dive
▸ Leaderless Replication — Deep Dive
Aspect
Single-Leader
Multi-Leader
Leaderless
Write path
All writes → 1 leader
Writes → nearest leader
Writes → ALL replicas (W of N)
Read scaling
Followers serve reads
Local leader + followers
Any replica serves reads
Write scaling
Bottleneck at leader
Distributed across leaders
No bottleneck (no leader)
Consistency
Strong (read from leader)
Eventual (conflicts possible)
Tunable (W+R>N = strong)
Conflicts
None (single writer)
Yes — LWW, CRDTs, custom
Yes — version vectors, CRDTs
Failover
Election needed (downtime)
Other leaders continue
No failover needed
Complexity
Simple
High (conflict resolution)
Medium (quorum tuning)
Best for
Read-heavy, single-region
Multi-region writes, collab
High availability, write-heavy
Examples
Postgres, MySQL, MongoDB, Redis
CouchDB, Postgres BDR, Google Docs
Cassandra, DynamoDB, Riak
Decision Guide: Need strong consistency + simple ops? → Single-Leader. Need multi-region low-latency writes? → Multi-Leader. Need maximum availability + no SPOF? → Leaderless. Most systems start single-leader and evolve as scale demands.
Sharding
Horizontal partitioning across independent database instances to improve
write scalability (distribute writes),
read scalability (parallel reads), and
storage capacity (split data),
with trade-offs in
complexity (routing, rebalancing),
cross-shard queries (expensive joins), and
hotspots (uneven key distribution).
▸ Sharding vs Replication
Sharding (Split Data)
Replication (Copy Data)
▸ Range vs Hash (Key-Based) Sharding
Range-Based Sharding
Key-Based (Hash) Sharding
▸ Directory & Geo Sharding
Directory-Based Sharding
Geo-Based Sharding
Strategy
How
Guarantee
Risk
Hash
hash(key) % N
Even distribution
Resharding requires data migration
Range
Split by value range (age, date)
Efficient range queries within shard
Hot spots for sequential keys
Directory
Lookup table maps key → shard
Flexible, any mapping
Lookup table = SPOF
Geo
Region-based (US→shard1, EU→shard2)
Data locality + compliance
Cross-region queries expensive
Real-world:Instagram shards Postgres by user_id (hash). Vitess (YouTube) — MySQL sharding middleware. Citus — distributed Postgres.
Consistent Hashing & Bounded Load
Map keys to nodes via a hash ring to improve
scalability (easy node add/remove),
load distribution (even spread with virtual nodes), and
stability (minimal key movement),
with trade-offs in
complexity (ring management) and
hotspots (skewed keys).
1 The Problem with Modulo Hashing
With 3 servers
server = hash(key) % 3
S1
S2
S3
K1→S1
K2→S2
K3→S3
K4→S1
K5→S2
K6→S3
Add 1 server (N=3 → 4)
server = hash(key) % 4
S1
S2
S3
S4
K1→S2
K2→S3
K3→S4
K4→S1
K5→S2
K6→S3
⚠ Almost all keys remapped → massive data movement, cache invalidation, latency spikes
2 Consistent Hashing — The Basic Idea
Hash servers onto a ring (0 to 2³²)
Hash keys onto the same ring
Each key goes to the first server clockwise
Key → Server Mapping
K1→S1K2→S1K3→S2K4→S3K5→S3K6→S1
S1 owns range (225, 75] S2 owns range (75, 150] S3 owns range (150, 225]
Only keys in a server's range belong to that server
3 What Happens When the Cluster Changes?
Add S4 at position 200
K5 was on S3 → now on S4 All other keys unchanged
Only keys in (150, 200] move from S3 → S4
Remove S2
K3 was on S2 → now on S3 (next clockwise) All other keys unchanged
Keys in (75, 150] move to next server clockwise
4 Why Virtual Nodes?
Without vnodes (3 servers, 1 point each)
A
60%
B
30%
C
10%
Uneven — A gets 6x more than C
With vnodes (each server × N points)
A
~33%
B
~33%
C
~33%
Even — many small ranges balance load
Virtual nodes = better balance + smoother failover + supports weighted capacity (bigger server → more vnodes)
5 How the Ring is Built (High Level)
Discover servers in cluster — get the list of active nodes
Generate virtual nodes — hash(serverId + i) for i = 1..N
Sort points on the ring — keep a sorted structure (e.g., array / skip list)
Handle replicas (optional) — pick next R distinct physical nodes on the ring for replication
6 Where is it Used?
System
How it Uses Consistent Hashing
Caching (Redis Cluster, Memcached)
Stable key-to-node mapping, minimal cache churn on scale
Databases / KV Stores (DynamoDB, Cassandra)
Sharding data with minimal resharding
Rate Limiting
Same user always hits the same limiter instance
CDN / Load Balancing
Same content routes to the same cache node
Message Brokers (Kafka)
Partition ownership with minimal movement
7 Bounded Load (Google, 2017)
Problem: Hot Keys
S1
20%
S2
15%
S3
65% 🔥
Even with vnodes, hash collisions or hot keys can overload one node
Fix: Cap at (1+ε) × avg
S1
30%
S2
32%
S3
38% ✓
If node is "full", key spills to next node on ring. No node exceeds cap.
✓ Key Takeaways
✓ Minimizes movement when nodes change ✓ Virtual nodes essential for even distribution ✓ Solves placement stability, not hot keys ✓ Topology consistency across clients is critical ✓ Think: minimize disruption under change
⚠ Watch Out For
⚠ Hot keys can still overwhelm a node ⚠ Data size imbalance (not just key count) ⚠ Different ring views → inconsistent routing ⚠ Too few virtual nodes → uneven load ⚠ Rebalancing is not free — do it intentionally
Mental Model: Consistent hashing is not about hashing. It's about stable key placement under change.
Bloom Filters
A probabilistic data structure that tells you: "Definitely not in the set" or "Probably in the set" (false positives possible, false negatives never). Uses tiny memory for ultra-fast checks at massive scale.
▸ Real-Time Example: "Is this username taken?" (Instagram/Twitter signup)
Why this matters at scale: Instagram has 500M+ usernames. Without Bloom filter: every signup checks DB (5ms × millions of signups/day = massive DB load). With Bloom filter: 99% of "is this taken?" checks answered in ~0.01ms from memory — only the 1% "maybe taken" results hit the DB. Filter size: 500M × 10 bits = ~600MB RAM (fits on one server).
▸ Lookup Process
Query: "bob" → Definitely NOT in set
Query: "user123" → Probably in set
Key Insight: If ANY bit is 0 → Definitely NOT present (zero false negatives). If ALL bits are 1 → Might be present (false positive possible). Memory: ~100KB for 1M items · O(K) insert + lookup.
▸ Why It Matters — Real-World
Cassandra / Bigtable
"Is key in this SSTable?" Check filter first → skip disk read if NO. Saves millions of I/O ops.
Chrome Safe Browsing
"Is this URL malware?" Check local bloom filter → only fetch from server if MAYBE. No network call for 99% of URLs.
Medium / Recommendations
"Has user read this article?" Filter per user → never recommend already-read content. Tiny memory per user.
CDN / Cache Check
"Is content cached at this edge?" Check filter → skip origin fetch if definitely not cached. Reduces origin load.
▸ Real-Time Example: Cassandra Read Path with Bloom Filter
▸ Multiple Hash Functions — Why K=3 or K=7?
Parameter
Symbol
Meaning
Typical Value
Impact
Bit array size
m
Total bits in the filter
~10 bits per item
More bits → fewer false positives, more memory
Number of items
n
Expected items to insert
Application-specific
More items → higher false positive rate
Hash functions
k
Number of independent hashes
k = (m/n) × ln(2) ≈ 7
Optimal k minimizes false positive rate
False positive rate
p
P(says "yes" when item absent)
~1% at m/n=10
Lower p → need more bits per item
Common Hash Functions Used
MurmurHash3: Fast, good distribution. Used by Cassandra, Guava. xxHash: Extremely fast (>10 GB/s). Used by Redis, LZ4. FNV-1a: Simple, decent distribution. Good for small filters. Double hashing trick: Use only 2 hash functions → derive K hashes as: h_i(x) = h1(x) + i × h2(x) (for i = 0..k-1) This gives K independent hashes from just 2 computations!
▸ Scaling Problem: What happens when data grows beyond initial m?
The problem: A standard Bloom filter has fixed size m (set at creation). You can't resize it — adding more bits would invalidate all existing hash positions. As n (items) grows beyond what m was designed for, the false positive rate degrades rapidly. Designed for 1M items but now have 10M? FP rate jumps from 1% to ~40%.
Solution 1: Scalable Bloom Filter
How: Chain multiple filters. When filter #1 fills up (FP rate exceeds threshold), create filter #2 with tighter FP rate (e.g., p/2). New inserts go to latest filter. Lookup checks ALL filters. Tradeoff: Lookup is O(K × number_of_filters) — slightly slower as it grows. Used by: Redis BF.RESERVE with expansion option.
Solution 2: Rebuild with larger m
How: When FP rate exceeds threshold, create a new filter with 2× the bits. Re-insert all items from the source of truth (DB). Swap atomically. Tradeoff: Requires access to original data. Downtime or dual-read during rebuild. Used by: Cassandra (rebuilds on compaction), periodic maintenance jobs.
Solution 3: Partitioned / Sharded Filters
How: Shard data by key range or hash. Each shard has its own Bloom filter sized for its subset. Add shards as data grows — each filter stays within its designed capacity. Tradeoff: Lookup must route to correct shard first. Used by: Cassandra (per-SSTable filter), HBase (per-region filter).
Practical rule: Size your initial filter for 2-3× expected items (headroom for growth). Monitor FP rate. When it crosses your threshold (e.g., >5%), trigger rebuild or chain a new filter. In Cassandra, this happens automatically during compaction — old SSTables merge → new Bloom filter built for the merged data.
Limitations:Cannot delete items (setting bit to 0 might affect other items). Cannot count (only membership). Cannot resize (must rebuild with larger m). No enumeration (can't list what's in the filter).
One-liner: Bloom filter = K hash functions mapping items to M bits. Insert: set K bits. Lookup: check K bits. Any 0 → definitely absent. All 1 → probably present. Use m/n=10, k=7 for ~1% FP rate. Real-world: Cassandra skips disk reads, Chrome blocks malware URLs, CDNs avoid origin fetches.
Rate Limiting
Throttle requests to improve
availability (prevent overload),
fairness (per-user limits), and
cost control (limit usage),
with trade-offs in
user experience (rejected requests) — return
429 Too Many Requests.
Fixed Window
Simple counter per window. Edge burst: 2× at window boundary.
Sliding Window
Weighted blend of prev + current window. No edge burst, memory efficient.
Token Bucket
Allows bursts up to bucket size. Most common (AWS, Stripe).
Leaky Bucket
Smooth output, no bursts. Queue absorbs spikes, fixed drain rate.
Algorithm
How It Works
Burst?
Memory
Accuracy
Best For
Real-World
Fixed Window → Simple but unfair
Counter resets every N seconds (e.g., every minute). Simple INCR + EXPIRE.
2× at edge — user sends max at end of window + start of next
Low — 1 counter per key
Approximate
Quick to implement, acceptable when occasional unfairness at window edges is fine
Twitter/X → 300 tweets/3hr Basic NGINX limit_req zone
Sliding Window Log → Fair and exact
Store timestamp of each request in sorted set. Count entries within last N seconds.
No
High — stores every request timestamp
Exact
Must guarantee exact count per user — no one gets even 1 extra request through
User identity unknown (pre-auth). Used for login, OTP, signup.
Per API Key
rate_limit:apikey:abc123
API Gateway
Stripe → Partner A: 1000/sec, Partner B: 100/sec
Each customer pays differently. Tied to billing tier.
Per Endpoint
rate_limit:endpoint:/generate
Application / Gateway
AI image gen → /profile: 1000/min, /generate: 10/min
Not all APIs cost the same. Expensive endpoints need stricter limits.
Global
rate_limit:service:total
Load Balancer / Gateway
DB protection → entire service max 50K req/sec
Protect backend from total overload regardless of who sends.
▸ 2. What's the Rule?
Single Rule
100 requests/minute Example: Weather API, public read endpoints Simple. One counter, one threshold.
Stacked Rules (Multi-Window)
10/sec + 100/min + 1000/hour Example: GitHub API Prevents both: short spikes (10/sec) AND long-term abuse (1000/hr). User can't send 100 instantly OR run forever at high rate.
▸ 3. What Happens When Limit Hits?
Response
Behavior
Example
When to Use
Reject (429)
Immediately return 429 Too Many Requests
GitHub, Stripe, public APIs
Simple, cheap, predictable. Client retries with backoff.
Queue
Accept request, process later from queue
Twilio SMS, email sending (100K emails queued)
User wants eventual delivery. Not time-critical.
Throttle
Degrade quality instead of failing
Netflix → reduce video bitrate instead of stopping
Better UX than hard rejection. Graceful degradation.
▸ 4. Tier-Based Limits
Limits tied to revenue:Free → 100 req/hr • Premium → 10,000 req/hr • Enterprise → 100,000 req/hr. Common in: SaaS APIs, AI platforms (OpenAI, Anthropic), cloud services (AWS). Store tier in user profile, look up at gateway.
▸ 5. How to Pick the Right Algorithm
NFR Question
If This...
Then Choose...
Why
Scale?
Small (100 users, internal)
Sliding Window Log
Exact, simple, memory is fine
Huge (10M+ users)
Sliding Counter or Token Bucket
Log stores every timestamp → memory explosion at scale
Single or Distributed?
One server
In-memory counter
No network hop needed
Multiple servers (Uber, Netflix)
Redis (shared state)
All nodes must see same count
Accuracy?
Security-critical (login, OTP)
Sliding Window Log
Even 2 extra attempts matter
Approximate fine (social posting)
Sliding Counter / Token Bucket
101 instead of 100 → not a disaster
Latency budget?
Ultra-sensitive (HFT, gaming)
Local in-memory limiter
Redis lookup adds 2-5ms
Normal APIs (e-commerce, SaaS)
Redis
2-5ms acceptable
Limiter store fails?
Fail-Open (allow traffic)
Public APIs, product catalog, news
Availability > protection. Risk abuse over downtime.
Fail-Closed (reject traffic)
Bank login, OTP, admin portal
Security > availability. Block all over unlimited attacks.
▸ 6. Implementation (Distributed — Redis)
Distributed challenges:Race conditions — two nodes check simultaneously. Fix: Redis Lua script (atomic). Clock skew — use Redis server time. Redis failure — fail-open (allow) for public APIs, fail-closed (block) for security-critical.
Anti-patterns:Rate limit only at app layer — DDoS bypasses. No Retry-After header — clients hammer blindly. Same limit for all endpoints — /login needs stricter limits than /search. No differentiation — paid users get same limits as free.
Interview tip:"I'd put rate limiting at the API Gateway layer using Redis sorted sets for a sliding window. Per-user limits with API key, per-IP limits for unauthenticated. Return 429 with Retry-After header. Fail-open on Redis failure to avoid blocking all traffic."
Backpressure
When the consumer can't keep up with the producer — the system must decide: slow down, buffer, or shed load
▸ Backpressure — Detection & Response Strategies
Strategy
Data Loss?
Latency Impact
When to Use
Example
Drop / Load Shed
Yes
None (fast)
Real-time, freshness > completeness
Live metrics, ad bids, video frames
Bounded Buffer + Block
No
Increases (producer waits)
Lossless, can tolerate delay
Reactive Streams, Go channels, Akka
Slow Producer (flow control)
No
Propagates upstream
End-to-end flow control
TCP window, Kafka quotas, HTTP/2
Scale Consumers
No
Temporary lag during scale-up
Elastic workloads
Kafka consumer groups, KEDA, Lambda
Sample / Aggregate
Partial
None
High-volume telemetry
StatsD (1-in-10), log sampling
Priority Queue
Low-priority dropped
None for high-priority
Mixed criticality traffic
Payment > analytics, VIP > free tier
▸ Backpressure Implementation by System
Network / Transport
TCP: sliding window shrinks when receiver is slow
HTTP/2: WINDOW_UPDATE frames control flow
gRPC: flow control per stream
WebSocket: application-level ACKs
Message Brokers
Kafka: consumer lag metric, fetch.max.bytes
RabbitMQ: prefetch count, credit-based flow
SQS: visibility timeout + DLQ
NATS: slow consumer detection, disconnect
Application Frameworks
Reactive Streams: request(N) protocol
Go channels: bounded channel blocks sender
Node.js streams: highWaterMark + drain event
Akka Streams: async boundaries with buffers
Key Insight:Always use bounded buffers. Unbounded queues are how outages turn into OOMs. Set buffer size based on: acceptable latency × throughput rate. Alarm on queue depth — growing depth means consumer is falling behind.
Monitoring: Track consumer lag (Kafka), queue depth (SQS/RabbitMQ), buffer utilization %, and drop rate. Set alerts at 60% capacity (warning) and 80% (critical). Use these signals to trigger auto-scaling.
Anti-patterns:Unbounded buffers — OOM guaranteed under sustained load. No backpressure signal — producer keeps flooding. Dropping without metrics — silent data loss. Blocking in async code — deadlocks the event loop.
Real-world:Netflix — RxJava backpressure for streaming pipeline. Uber — QALM (queue-based adaptive load management). Twitter — Finagle admission control rejects when overloaded. Kafka — consumer lag triggers auto-scaling via KEDA.
Auto-Scaling
Match capacity to load — reactively (metrics-driven) or predictively (ML/schedule-based). Scale up fast, scale down slow.
▸ Auto-Scaling — Signals, Decisions & Actions
Platform
Tool
Signals
Key Feature
Kubernetes
HPA (Horizontal Pod Autoscaler)
CPU, memory, custom metrics
Native, works with any metric via Prometheus adapter
Horizontal: add more instances (preferred, no ceiling)
Vertical: bigger instance (simpler, has ceiling)
Stateless services → horizontal easy
Stateful (DB) → vertical first, then shard
Combine: vertical for DB, horizontal for app tier
Cold Start & Warm Pools
Cold start: new instance needs boot + warm cache
Warm pools: pre-initialized instances ready to serve
Predictive: ML forecasts traffic, pre-scales
Scheduled: scale before known peaks (9am, sales)
Lambda cold start: 100ms-10s depending on runtime
Combine signals: Base on CPU for steady state, but use queue depth for burst detection (leading indicator). Add predictive scaling to pre-warm before known peaks. Set min instances to handle baseline + sudden spikes during scale-up lag.
Scale-down safety: Use connection draining (deregister from LB, finish in-flight requests). Set termination grace period (K8s: preStop hook). Scale down 1 at a time to avoid cascading failures. Never scale below N+1 capacity (survive 1 instance failure).
Anti-patterns:Scaling on a single metric — CPU high but memory is the bottleneck. No cooldown — thrashing (scale up/down/up/down). No max limit — runaway scaling = runaway cost. Scaling stateful services without draining — data loss.
Real-world:Netflix — Titus auto-scaling with predictive models (pre-warm for evening peak). Uber — Peloton scheduler scales 4M+ containers. Shopify — scales to 80K+ RPS during flash sales using predictive + reactive. Slack — KEDA scales workers based on job queue depth.
Graceful Degradation
Stay partially up instead of fully down — shed non-critical features to protect core user journeys
Tactics:Feature flags (LaunchDarkly) for instant toggle. Traffic tiering by importance (paid > free). Cached fallbacks with extended TTL. Circuit breakers per dependency. Request hedging for tail latency. Bulkhead isolation — failure in one component doesn't cascade.
Testing degradation:Chaos engineering — inject failures in prod (Netflix Chaos Monkey). Game days — simulate outages with the team. Load testing — push past capacity to verify degradation tiers activate correctly. Dependency kill switch — test each circuit breaker independently.
Anti-patterns:All-or-nothing — either fully up or fully down (no middle ground). No fallback tested — fallback path has its own bugs. Cascading failures — one service down takes everything down. Manual intervention required — degradation should be automatic.
Real-world:Netflix — Hystrix circuit breakers + fallback to cached recommendations. Amazon — static product pages from S3 when app tier is overloaded. Twitter — fail whale → degraded timeline (no trends, no who-to-follow). Shopify — checkout always works, even if storefront is degraded during flash sales.