HDFS stores the bulk events and MinIO serves the photos, but neither of them can answer "Anna's latest orders" in a few milliseconds, or deduct two units of aged-cheese without selling the same piece twice. That is what databases are for and, when a single machine is no longer enough, distributed databases. This lesson brings together everything that Module 3 and the previous lessons have prepared: the partitioning and consistent hashing of 04-01, the replication and quorums of 03-04, the consistency models and the CAP table of 03-02, and the transactions of 03-05. We will look at the three paths that exist for distributing a database (scaled-out relational, Dynamo-style NoSQL and document), compare them, and take the most important decision of the module for Kilometre Zero: orders migrates to Cassandra and inventory stays on PostgreSQL. Then we will make it real: Cassandra in docker-compose.yml, the km0_orders keyspace, the orders_by_customer and orders_by_id tables, and a Python repository that chooses the consistency level on every operation.

1. What a distributed database is, and its transparencies
2. Horizontal and vertical fragmentation
3. Path (a): the scaled-out relational database and NewSQL
4. Path (b): Dynamo-style NoSQL with Cassandra
5. Query-driven modelling in Cassandra
6. Path (c): document databases with MongoDB
7. Comparison table
8. Kilometre Zero's decision
9. Hands-on: Cassandra in docker-compose.yml and cassandra_repository.py
10. Common Mistakes and Tips
11. Exercises
12. Conclusion

A distributed database is a collection of logically related data, physically spread across several nodes connected by a network, that presents itself to applications as a single database. The definition has two halves: the physical distribution (which we already know how to do with partitions and replicas) and the illusion of unity, which is once again the list of transparencies from 01-01 applied to data:

The last row is the one that separates the paths. Maintaining transaction transparency across nodes requires 2PC or consensus (03-03, 03-05) on every write that crosses partitions, and each path decides how much of that it allows.

In the classic vocabulary of distributed databases, partitioning a table is called fragmenting it:

• Horizontal fragmentation: each fragment holds a subset of the rows, with all the columns. It is exactly the partitioning of 04-01 (by range, hash or compound) applied to a table, and in NoSQL jargon it is called sharding. Anna's orders on one node, Mark's on another.
• Vertical fragmentation: each fragment holds a subset of the columns (with the primary key repeated). An order's billing columns on one node, its delivery columns on another. It is uncommon between nodes of a single database, but it is precisely what Kilometre Zero did when it split the monolith: orders, inventory, payments and delivery are vertical fragments of the old schema, each owning its columns, with order_id as the shared key and no joins between services (01-06).
• Mixed fragmentation: vertical between services, horizontal within each one. This is the platform's actual situation: km0_orders is a vertical fragment which is in turn partitioned horizontally by customer_id.

Three rules of good fragmentation (Özsu and Valduriez): completeness (every row or column is in some fragment), reconstruction (the original table can be put back together with unions) and disjointness (a row is in only one fragment, except for the key in the vertical case). Replication deliberately relaxes the third.

The first path keeps SQL, the relational model and ACID transactions, and adds distribution in layers:

Read replicas. We set this up in 03-04: orders-db-primary receives all the writes and orders-db-replica (and any others that are added) serve reads via streaming replication. It scales reads, not writes or size, and it brings eventual consistency on the replicas with it (reading your own write means going to the primary or waiting for the LSN).

Application-level sharding. When writes or volume overwhelm a single primary, the application spreads the rows over several independent PostgreSQL instances using the logic of 04-01: shard = ring.node_for(customer_id), one connection per shard, and the application knows which shard each piece of data is on. It is what Instagram, Notion or Shopify did for years. It works, but fragmentation transparency disappears: joins between shards are done in the application, transactions across shards do not exist (or are sagas, 03-05), rebalancing shards is a project in its own right, and every new use case has to respect the partition key.

Citus. A PostgreSQL extension that automates that sharding: a coordinator node receives ordinary SQL queries, and "distributed" tables (SELECT create_distributed_table('orders', 'customer_id')) are spread by hash across worker nodes. The coordinator rewrites each query into per-shard subqueries and combines the results; transactions that touch a single shard are local, and those that touch several use 2PC between workers (03-05, with its latencies). It is the middle ground: SQL and PostgreSQL with automatic distribution, ideal for multi-tenant workloads where almost every query includes the distribution key.

NewSQL. Databases designed from scratch to be distributed without giving up SQL or ACID:

NewSQL is CP with transactions (03-02: it chooses consistency under a partition, and consistency over latency in normal operation). What you pay is latency per write (one consensus round per range, plus 2PC if there are several) and operational complexity. It is the right answer when you need cross-partition transactions and full SQL at scale; it is not needed for most of Kilometre Zero's services, which have already given up cross-domain transactions in favour of sagas.

In 2007 Amazon published the design of Dynamo, its key-value store for the shopping basket, which optimised for availability and write latency: leaderless, consistent hashing, configurable quorums, version vectors and read repair. Cassandra (Facebook, 2008; now Apache) combined Dynamo's architecture with Bigtable's column-family data model. Riak, Voldemort and ScyllaDB belong to the same family. Everything that follows is 04-01 and 03-04 put into practice:

A consistent hashing ring. Each node owns ranges of tokens on the Murmur3 ring (−2⁶³ to 2⁶³−1); with num_tokens: 16 (Cassandra 4) each node has 16 vnodes. The partition key is hashed, and the resulting token determines the coordinator node for that partition and its replicas: the next N−1 distinct positions on the ring (skipping vnodes of the same node and, with NetworkTopologyStrategy, spreading them across racks and data centres).

Leaderless. Any node accepts any request (the "any node" option from 04-01): the node that receives it acts as the coordinator, forwards the write to the partition's N replicas and waits for as many acknowledgements as the consistency level demands. There is no leader election and no failover: if a replica is down, the others keep accepting writes and the coordinator stores a hinted handoff (03-04) to deliver to it when it comes back.

Replication factor and strategy. These are set per keyspace: SimpleStrategy (replicas on the next nodes round the ring, for testing only) or NetworkTopologyStrategy with a factor per data centre ({'dc-bcn': 3, 'dc-vlc': 3}), which also spreads the replicas across racks. Changing the factor requires a nodetool repair so that the new replicas get populated.

Per-operation consistency levels. This is Cassandra's most elegant contribution: every read and every write chooses how many replicas must respond. With N = 3:

It is exactly the W + R > N of 03-04 with quorum_wr.py, but decided on each operation: the position of van-3 is written with ONE (fast, AP), the confirmation of an order with QUORUM (consistent, CP), and the read that shows Anna her newly confirmed order with QUORUM so that she sees what she has just written. Reads perform read repair when replicas disagree, and nodetool repair (anti-entropy with Merkle trees) reconciles in the background.

Writing to disk. A write goes to the commit log (sequential, durable) and to the in-memory memtable; when the memtable fills up, it is flushed to an immutable SSTable on disk. SSTables are periodically compacted, merging versions. This design (the LSM-tree) makes writes sequential and extremely cheap, which is why Cassandra is the classic choice for write-heavy workloads; reads may have to consult several SSTables (mitigated with Bloom filters and caches).

Tombstones. Because SSTables are immutable, deleting does not delete: it writes a timestamped tombstone that hides the value. The tombstone is removed at compaction after gc_grace_seconds (10 days by default), a period that must be longer than it takes to repair any replica that was down; otherwise, the downed replica would "resurrect" the deleted data when reconciling. Lots of tombstones in a partition (mass deletes, queues implemented on Cassandra) seriously degrade reads: it is Cassandra's most famous anti-pattern.

In SQL you model the domain (normalised) and then write whatever queries you need, with joins and indexes. In Cassandra there are no joins and no efficient ad hoc queries, so the order is reversed: queries first, then one table per query, denormalising as much as necessary. The primary key has two parts:

• The partition key (customer_id) (the inner parentheses) determines the token and therefore the node: all of Anna's orders are together, on the same node (and its replicas). It is the decision from 04-01, section 10, turned into a schema.
• The clustering columns created_at, order_id sort the rows within the partition, on disk. CLUSTERING ORDER BY (created_at DESC) stores the most recent first, so "Anna's latest 20 orders" means reading the first 20 rows of the partition: a single seek, a single node.
• The efficient queries are those that fix the complete partition key and, optionally, a range over the clustering columns in order: WHERE customer_id = 'anna', WHERE customer_id = 'anna' AND created_at > '2026-09-01'. A query without the partition key (WHERE status = 'PENDING') requires ALLOW FILTERING and scans the whole cluster: it is the scatter/gather of 04-01 in its worst form, and it is banned in production.
• For "order by id" (the confirmation page, the link in the email) you need another table, orders_by_id, with PRIMARY KEY ((order_id)), and the application writes to both in a logged BATCH (atomic across the two tables, though not isolated). That is the price of denormalisation, and it is cheap because writes are.
• Partitions must be kept bounded (recommendation: < 100 MB and < 100,000 rows): a restaurant customer with tens of thousands of orders would break the rule, which is why 04-01 proposed the monthly bucket (customer_id, year_month) as a composite partition key. We will apply it as an exercise.

Document databases store JSON/BSON documents with a flexible schema and rich queries over any field, with secondary indexes. MongoDB distributes them at two levels:

• Replica set: one primary and several secondaries with asynchronous replication via the oplog and automatic election of the primary (a protocol derived from Raft, 03-03). The application chooses the write concern (w: 1, w: "majority") and the read preference (primary, secondaryPreferred), which are W and R again under different names. It is leader-follower with failover (03-04).
• Sharding: collections are split into chunks by range or hash of a shard key, each chunk lives on a replica set, the mongos routers route the queries (the "routing tier" option from 04-01), and the config servers (a replica set) store the chunk map, which a balancer moves around to rebalance. Queries that include the shard key go to one shard; without it, to all of them.
• Transactions: always ACID within a document (an order with its embedded lines is written atomically); since version 4.0/4.2, multi-document and multi-shard transactions with internal 2PC, slower and with limits.

It fits when the domain is naturally expressed as self-contained documents with varied queries (catalogues, profiles, content). We cover it briefly because, for Kilometre Zero, the catalogue (one document per product with variants and photos) would be a natural candidate, but the team decided to keep it in PostgreSQL with jsonb columns (which cover 90% of the document use case) plus the Redis cache (04-05): one fewer database to operate.

With this table in front of them, along with the CAP decision table from 03-02, the team decides service by service:

orders migrates to Cassandra. Reasons:

1. Volume and write-heavy load: 40,000 orders/day during a campaign with 6 events per order, plus the full history (the law requires it to be kept for years) and the sagas table (03-05) with one transition per step. It is a workload of sequential writes and reads by known key, the strength of the LSM-tree and of the leaderless ring.
2. Known, stable queries: "my orders" (by customer, sorted by date), "order by id", and the indexes by market and producer materialised from orders.events. None of them needs joins or online aggregations (those go to the data lake and Module 5).
3. Availability: creating an order must work even if a node, or even a data centre, fails; with LOCAL_QUORUM and RF=3 per DC, the write goes through with one node down. The order no longer needs transactions across services (sagas) or across tables beyond a batch, so per-partition atomicity is enough.
4. Linear scale: adding nodes adds proportional capacity, and 04-01 showed that the ring with vnodes moves only what is strictly necessary.

inventory stays on PostgreSQL (the primary + replica from 03-04, partitioned by product_slug when it becomes necessary). Reasons:

1. CP counters: stock is a constraint (CHECK (available >= 0)) that has to be enforced on every reservation, with UPDATE … WHERE available >= 2 locking the row. Cassandra has no constraints, its counters are neither idempotent nor transactional, and lightweight transactions (Paxos per operation) would be far slower than an UPDATE in PostgreSQL.
2. Modest volume: 4,200 products times 4 markets is about 17,000 stock rows; the problem is not size but correctness and contention on the hot keys, which will be tackled with the cache and with queues (04-05), not with more nodes.
3. Local transactions: reserving stock and recording the idempotent reservation (processed_messages, 02-05) in the same transaction, with the outbox table (02-05) in the same database: PostgreSQL does it in one COMMIT.

payments also stays on PostgreSQL (CP, low volume, auditing); catalog on PostgreSQL with jsonb and Redis in front; delivery (the positions of van-3, 2.4 million a day, AP) is a candidate for Cassandra with daily partitions per courier, as exercise 2 of 04-01 suggested; analytics in the lake (04-02) and in an analytical store that Module 5 will choose.

Three Cassandra nodes in a single data centre (dc-bcn), with two simulated racks, so that we can use NetworkTopologyStrategy and see how the replicas are distributed:

The nodes must start one at a time (hence the dependencies): two nodes joining the ring at once with the same seed is one of the classic causes of a badly formed cluster. Start them up and check the ring:

UN = Up, Normal. Owns (effective) is 100% on every node because, with RF=3 and 3 nodes, each node holds a replica of everything; with 6 nodes we would see ~50%. nodetool ring shows the ring's 48 tokens (16 per node), and nodetool getendpoints km0_orders orders_by_customer anna will tell you which three nodes store Anna's partition.

The schema, using cqlsh:

The line type is a UDT (user-defined type), and frozen means the list is stored as a single serialized value (individual elements cannot be modified, but it is read and written in one go, which is what we want). orders_by_customer and orders_by_id hold the same data under different keys: this is the denormalisation of section 5.

The Python repository, using cassandra-driver (the official DataStax driver; pip install cassandra-driver):

What is worth understanding about the code:

• TokenAwarePolicy(DCAwareRoundRobinPolicy): the driver downloads the ring's token map and sends each request directly to a replica of the partition (the "partition-aware client" option from 04-01), avoiding the coordinator's extra hop; and it prefers nodes in the local data centre.
• ExecutionProfile sets LOCAL_QUORUM as the default, and each statement can override it with consistency_level. This is the central point of the exercise: consistency is a property of the operation, not of the database. latest_orders uses ONE on the normal path (the order list can tolerate a few milliseconds of lag) and LOCAL_QUORUM when the website calls it right after create_order, which was also LOCAL_QUORUM: 2 + 2 > 3, so the read sees the write. It is the read-your-writes model of 03-01 achieved with quorums.
• BatchStatement(LOGGED): Cassandra first writes the batch to a replicated batchlog and guarantees that both inserts will end up being applied (eventual atomicity, with no isolation: a reader may see one table updated and not the other for a few milliseconds). Batches are for keeping denormalised tables coherent, not for "going faster": a batch spanning hundreds of different partitions is an anti-pattern.
• Prepared statements with ?: they are sent to the server once and reused with parameters, and the driver knows which parameter is the partition key for token-aware routing.
• change_status shows the cost of denormalisation: to update orders_by_customer you need the complete key (customer_id, created_at, order_id), which is obtained from orders_by_id. In the saga of 03-05, the orchestrator already knows this data and saves itself the read.

Run the script, and then try out the levels with the cluster degraded:

With one node down, QUORUM keeps reading and writing (2 live replicas out of 3) and ALL answers Unavailable immediately: availability and consistency chosen operation by operation, as 03-02 promised. When cassandra-3 comes back, the hints accumulated by cassandra-1 and cassandra-2 are delivered to it, and nodetool repair km0_orders reconciles any remaining difference.

• Modelling Cassandra like SQL. A normalised orders table with secondary indexes for every query ends in ALLOW FILTERING and scatter/gather. List the queries, one table per query, denormalise with batches.
• Unbounded partitions. A customer, a courier or a product with millions of rows in the same partition degrades compactions and reads. Add a time bucket to the partition key.
• Using Cassandra for queues or for mass deletes. Tombstones pile up and reads die. If you need a queue, you already have Kafka (02-04).
• SimpleStrategy in production. It knows nothing about racks or data centres: three replicas in the same rack. Always NetworkTopologyStrategy, even with a single DC.
• gc_grace_seconds shorter than the maximum repair time. A node that comes back later resurrects deleted data. Repair every node within that window (a scheduled nodetool repair).
• Choosing NewSQL or Cassandra "because it scales" with 20 GB of data. A well-indexed PostgreSQL on a big machine serves tens of thousands of transactions per second. Distribute when volume, writes or availability demand it, and only then will you pay for rigid modelling or consensus latency.
• Business counters in Cassandra. counter columns are not idempotent (a retry doubles the increment) and do not support conditions. Stock, balances and anything with a constraint go to PostgreSQL or NewSQL.
• Ignoring the driver's default consistency level. In cassandra-driver it is LOCAL_ONE. Set it explicitly in the ExecutionProfile and decide per operation.

Exercise 1. Apply the bounded-partition recommendation: redefine orders_by_customer with the composite partition key ((customer_id, year_month), created_at, order_id), where year_month is a text value such as '2026-09'. Write the CREATE TABLE, adapt create_order to compute year_month from created_at, and rewrite latest_orders(customer_id, limit) so that it returns the latest 20 orders even if they are spread over several months (hint: walk backwards through the months until the limit is reached or a cap is hit). What happens with a customer who has not bought anything in the last 12 months?

Exercise 2. For each Kilometre Zero operation, choose the Cassandra consistency level (with RF=3 in dc-bcn and RF=3 in dc-vlc) and justify it with W + R > N and with the CAP table from 03-02: (a) writing a position of van-3; (b) reading the latest position for the customer-facing website; (c) confirming order P-2026-000126 on receiving payment.confirmed; (d) reading that order from the confirmation page 200 ms later; (e) the nightly analytics report that reads all of the day's orders; (f) reserving a new producer's username, which must be unique.

Exercise 3. A colleague proposes migrating inventory to Cassandra as well "so as to have a single database", implementing stock as UPDATE stock SET available = available - 2 WHERE product = 'aged-cheese' with a counter type, and then checking with a SELECT that it has not gone negative. Explain with a concrete scenario (two concurrent reservations by Anna and Mark for the last 3 pieces) why it fails, what alternative Cassandra offers (lightweight transactions with IF) and why, even so, the decision in section 8 stands.

Solution 1:

Each iteration is a read of a different partition (one partition per month), so "the latest 20" costs between 1 and max_months reads, almost always 1 or 2 for an active customer. A customer with no purchases in 12 months gets an empty list after 12 quick reads (non-existent partitions are resolved by Bloom filters without touching disk): the cap avoids walking back through years, and if the business needs the full history you add an explicit query by month range or a summary table months_with_orders_by_customer.

Solution 2:

Solution 3:

Scenario: 3 pieces are left. Anna reserves 2 and Mark reserves 2 at almost the same time. With counter, both UPDATEs are applied unconditionally: the counter goes to 3 − 2 − 2 = −1. The follow-up SELECT checks both see −1, and each one tries to "undo" by adding 2: the counter ends up at 3, or 1, or −1 depending on the interleaving, and neither knows whether its reservation is valid. Worse: if a counter UPDATE is retried because of a timeout (03-05, retriable sagas), it is applied twice, because counters are not idempotent. The correct alternative in Cassandra is a lightweight transaction: UPDATE stock SET available = 1 WHERE product = 'aged-cheese' IF available = 3, which runs Paxos among the partition's replicas and returns [applied] = false to whoever arrives second, who must re-read and retry (compare-and-set). It works, but every reservation costs four network round trips between replicas (a few tens of milliseconds) compared with an UPDATE … WHERE available >= 2 in PostgreSQL with a row lock (less than a millisecond), and with no declarative constraints and no transaction with the processed_messages table and the outbox. With aged-cheese as a hot key during Artisan Cheese Week, contention in Paxos would be the bottleneck. That is why inventory stays on PostgreSQL: the problem is the correctness of a contended counter, not volume, and that is the strength of a CP relational database.

A distributed database physically spreads the data and offers the transparencies of fragmentation, replication, location, failure and, to varying degrees, transaction. Horizontal fragmentation is the partitioning of 04-01 applied to rows, and vertical fragmentation is what Kilometre Zero did when it shared out the monolith's schema among services. There are three paths: the scaled-out relational database with read replicas, application-level sharding or Citus, and in its most ambitious version NewSQL (Spanner with TrueTime, CockroachDB and YugabyteDB with Raft), which keeps global ACID at the cost of latency; the Dynamo-style NoSQL that Cassandra embodies with a leaderless consistent hashing ring, a per-keyspace replication factor and a consistency level chosen on every operation (ONE, QUORUM, ALL) as a literal application of W + R > N, with query-driven modelling (one table per query, partition key plus clustering columns) and with tombstones as its toll; and document databases such as MongoDB with replica sets and sharding. Kilometre Zero has decided that orders will migrate to Cassandra because of its volume, its write-heavy load and its stable queries, and that inventory will stay on PostgreSQL because a counter with a constraint is a CP problem of correctness, not of scale. We have set it up with three nodes in docker-compose.yml, the km0_orders keyspace with NetworkTopologyStrategy, the orders_by_customer and orders_by_id tables, and cassandra_repository.py writing in a batch and reading with ONE or LOCAL_QUORUM according to what each operation needs, checking with nodetool status and one node stopped that QUORUM survives and ALL does not.

Almost all of Kilometre Zero's data now has its place. But the catalogue is queried hundreds of times for every time it changes, and every query that reaches PostgreSQL during Artisan Cheese Week is a query the database might not have had to answer. The last piece of the module sits between the application and the databases: distributed caches, with Redis, its patterns, its classic problems and a Redis Cluster whose 16,384 slots are the latest reincarnation of the partitioning with which the module began.