Distributed Data Management

One machine cannot hold all your data and one machine cannot survive forever. Distributed data management is the discipline of splitting state across many machines (sharding) and keeping multiple copies of each piece (replication) so that the system stays available, durable, and fast enough — while exposing a coherent enough story to applications that they can be written without thinking about every node.

Replication Strategies

Three classic models, each a different trade-off:

• Single-leader (PostgreSQL streaming, MySQL, MongoDB primary): all writes go through one leader. Followers replicate the leader's log. Simple to reason about, but the leader is a bottleneck and a SPOF (mitigated by automatic failover).
• Multi-leader (multi-region MySQL with bidirectional replication, CRDT-backed systems): writes accepted at multiple nodes; conflicts resolved by merge rules. Harder; useful for geo-distributed systems where local writes matter.
• Leaderless (Cassandra, DynamoDB, Riak): writes are sent to multiple replicas in parallel; reads are reconciled via quorum. No single “leader” per partition.

Synchronous vs asynchronous replication is orthogonal:

• Synchronous: write is acknowledged after at least one replica confirms. Higher latency, no data loss on leader crash.
• Asynchronous: write is acknowledged on leader commit; replicas catch up later. Lower latency; potential data loss window.
• Semi-synchronous: hybrid; ack after any one replica confirms.

Sharding Strategies

Hash partitioning: partition = hash(key) % N. Excellent load distribution; range queries are expensive (every shard touched). Used by Cassandra, DynamoDB, Redis Cluster.

Range partitioning: each shard owns a contiguous key range. Range queries are efficient; hot spots are easy to create (a key prefix that is heavily written becomes a single-shard bottleneck). Used by HBase, BigTable, CockroachDB, MongoDB sharded clusters.

Consistent hashing: solves the “adding a node forces all keys to move” problem of naive hash partitioning. Each node is hashed onto a ring; each key belongs to the next node clockwise. Adding a node only moves 1/N of keys. Combined with virtual nodes (256 vnodes per physical node in Cassandra) for smoother rebalancing.

The Distributed Systems Algorithms guide goes deeper on consistent hashing math.

Quorum Math

With N replicas, write quorum W, and read quorum R, you achieve strong consistency when W + R > N. The intuition: any read overlaps any write by at least one node, so the latest write is visible.

• N=3, W=2, R=2: tolerates 1 failure for both reads and writes; strong consistency. Standard Cassandra production.
• N=3, W=3, R=1: every write hits all replicas; reads are fast and consistent; one failure blocks writes. Read-heavy workloads.
• N=3, W=1, R=1: maximum availability and lowest latency; eventual consistency only. DynamoDB default.
• N=5, W=3, R=3: tolerates 2 simultaneous failures; strong consistency. Mission-critical Cassandra clusters.

Cassandra exposes this as consistency levels (ONE, QUORUM, LOCAL_QUORUM, EACH_QUORUM, ALL). For multi-region deployments LOCAL_QUORUM is critical — it requires a quorum within the local datacentre but does not wait for cross-region acks.

Consistency Models in Practice

The same hierarchy you saw in Module 1 (Linearizable → Sequential → Causal → Read-your-writes → Eventual) shows up in every real database, often as tunable guarantees:

• Cassandra: per-query consistency level.
• MongoDB: readConcern (local, majority, linearizable) + writeConcern.
• DynamoDB: per-call ConsistentRead flag.
• Spanner / CockroachDB: linearizable by default, with explicit follower-read modes.

Distributed Database Choices

• Postgres / MySQL: single-leader replication; horizontal scale via read replicas or external sharding (Citus, Vitess). Strong consistency on the leader; read lag on replicas.
• Cassandra: leaderless, hash-partitioned, AP-leaning. High write throughput, tunable consistency. Great for write-heavy time-series; less great for low-latency reads of small data.
• DynamoDB: managed, hash-partitioned, AP-leaning. Eventual or strong consistency per call. Excellent if you stay within its access patterns; expensive if you fight them.
• CockroachDB: distributed SQL with per-range Raft consensus. Linearizable by default; horizontal scale; SQL surface. The heaviest of the modern options.
• Spanner / TiDB: globally-distributed SQL with linearizable transactions backed by TrueTime / TSO. Premium pricing for premium consistency.
• Redis Cluster: in-memory, hash-slot partitioned, asynchronous replication. Great cache or session store; not your primary database.

Common Pitfalls

• Read-your-writes anomaly: writing to leader, reading from a replica, getting your own old value. Fix: stickify reads to leader for a short window after a write.
• Hot partition: a partition key with skewed traffic (one tenant, one celebrity user) overloads one node. Fix: salted keys, increased shard count, multi-tenant isolation.
• Cross-shard transactions: expensive (2PC, distributed locking). Design data models to keep related data in the same shard wherever possible.
• Replication lag spikes: replicas fall behind during bulk writes; reads served from lagging replicas return stale data. Monitor lag in seconds or bytes; fail reads to leader past a threshold.

Quorum Write Flow

Self-Check Quiz

1. N=5 Cassandra cluster. You write with CL=QUORUM, then immediately read with CL=ONE. Can you see your write? (Answer: not guaranteed. CL=ONE returns the first replica response, which may be a stale one. For strong consistency, you need W+R>N — e.g. CL=QUORUM on both.)
2. Your single celebrity user generates 90% of writes for one shard. What is happening, and what is the fix? (Answer: hot partition. Fix: split the key with salting (user_id:0..N) and aggregate at read time, or scale out and use a different partitioning key.)
3. Why does range-partitioned MongoDB sometimes accidentally create hot shards while hash-partitioned Cassandra rarely does? (Answer: range partitioning concentrates time-prefix or sequential keys on one shard; hash partitioning randomises. Choose the partition strategy based on access pattern.)
4. You see replication lag of 30 seconds on a Postgres read replica. What three actions matter most? (Answer: alert if it exceeds threshold; route latency-sensitive reads to leader; investigate root cause — bulk writes, slow disk, or vacuum.)

For the algorithm-level treatment of consistent hashing, vector clocks, CRDTs, and quorum proofs, read the Distributed Systems Algorithms guide. For Kubernetes-specific operational patterns the Kubernetes cheatsheet is the fast reference.