Break
Distributed systems
Nano quiz ↗
press b or esc to carry on
M5

Sharding vs Replication

*Sharding* splits one logical dataset across many machines so the cluster can hold and serve more data than any single node can.

3 figures
sharding-replication

Sharding vs replication: capacity vs availability

Sharding vs replication: capacity vs availability Two panels. Left, sharding: a 10 TB dataset splits into ten disjoint shards, about 1 TB each, one per node, so total capacity scales past one machine. Right, replication: one shard is copied onto three machines so any single node can fail without losing data, which buys availability, not extra capacity. The two use distinct counts on purpose: ten shards for capacity, three replicas for safety. Color key: pink is the mechanism in focus, grey is the data and the machines. Sharding vs replication Different jobs: 10 shards scale capacity, 3 replicas survive failure. Sharding 10 TB split into 10 disjoint shards S1S2S3S4S5 S6S7S8S9S10 10 nodes, about 1 TB each Each row on one shard · capacity scales past one machine Replication One shard copied to 3 machines 1 shard ~1 TB Copy on Node A Copy on Node B Copy on Node C Any one node can fail Availability, not extra capacity Color key  pink = the mechanism in focus  ·  grey = the data and the machines
Sharding and replication solve different problems and almost always run together. Sharding splits a 10 TB table into disjoint slices across machines, each row on exactly one shard, to scale capacity past one machine. Replication keeps two or three copies of the same data on different machines to survive single-machine failures and serve more reads. Production systems shard then replicate each slice, taking capacity from sharding and availability from replication, as Cassandra, sharded MongoDB clusters, and BigQuery's storage layer do.
Notes ↗
sharding-replication · Production Synthesis

In production: shard, then replicate

In production: shard, then replicate Production runs both mechanisms together. Spotify's 10 TB Listens table is sharded ten ways for capacity, then each shard is replicated three times on different racks for safety, so a user's shard sits on three machines and any one can serve the read. The figure shows the ten shards, one shard fanned out to three replicas, and a read served by any replica. Color key: pink is the shard in focus and its three replicas, grey is the other shards and the machines. In Production: Shard, Then Replicate Spotify's 10 TB Listens: 10 shards for size, 3 replicas each for safety. Sharded 10 ways — capacity S1S2S3S4S5S6S7S8S9S10 S3 Each shard → 3 replicas on different racks S3 copy · Rack 1S3 copy · Rack 2S3 copy · Rack 3 healthyhealthyhealthy A read hits any one replica. A user's shard sits on 3 machines, so a dead machine loses nothing. Color key  pink = the shard in focus & its 3 replicas  ·  grey = the other shards & machines
In production, both at once. Spotify's 10 TB Listens is sharded ten ways for capacity, then each shard is replicated three times on different racks. A user's shard lives on three machines, so a read can hit any of them and a dead machine loses nothing.
Notes ↗
sharding-replication · Common Confusions

Why you need both: each mechanism alone leaves a gap

Why you need both: each mechanism alone leaves a gap Two failure modes. Left, sharding without replication: ten shards on ten machines, but when one machine dies its shard is offline, locking out one in ten users. Right, replication without sharding: every machine holds the whole 10 TB dataset, which does not fit on a 1 TB disk. Only running both together avoids both gaps. Color key: red is the failure, pink is the data at risk, grey is the machines. Why You Need Both Each mechanism alone leaves a gap. Sharding without replication 10 shards, one copy each S1S2S3S4S5S6S8S9S10 S7 gone One machine dies → its shard is offline 1 in 10 users locked out Replication without sharding Every machine holds the whole dataset Node ANode BNode C 10 TB10 TB10 TBon 1 TBon 1 TBon 1 TB 10 TB will not fit on a 1 TB disk Copies do not add capacity Color key  red = the failure  ·  pink = the data at risk  ·  grey = the machines
Why both are needed. Sharding without replication (left): when a machine dies its shard goes offline, locking out one in ten users. Replication without sharding (right): every machine must hold the whole 10 TB, which never fits on a 1 TB disk. Only running both closes both gaps.
Notes ↗
M5

Distributed Commit

When one transaction touches several shards, *two-phase commit* makes the commit decision atomic across all of them: a coordinator first asks every shard to *prepare*, then tells them all to *commit* only if every shard voted yes.

1 figure
distributed-commit · Two-Phase Commit

Two-phase commit: prepare, then commit

Two-phase commit: prepare, then commit Two panels. Left, phase one (prepare): a coordinator asks every shard whether it can commit; each shard writes a log record and votes yes, so the coordinator's decision becomes commit. Any single no would make the decision abort. Right, phase two (commit): the coordinator tells every shard to commit and each acknowledges, so the multi-shard transaction is atomic, all commit or none do. Color key: pink is the coordinator's decision, grey is the shards and the messages between them. Two-phase commit: prepare, then commit One transaction spanning many shards commits atomically, or not at all. Phase 1 · Prepare Coordinator asks; each shard votes Coordinator asks all Shard 1 · log + vote YES Shard 2 · log + vote YES Shard 3 · log + vote YES All YES → decision = COMMIT any single NO → decision = ABORT Coordinator logs the decision before telling anyone Phase 2 · Commit Coordinator tells all; each acknowledges Coordinator says COMMIT Shard 1 · commit + ack Shard 2 · commit + ack Shard 3 · commit + ack Atomic: all commit, or none do no shard can disagree with the decision Coordinator crash after prepare blocks the shards Color key  pink = the coordinator's decision  ·  grey = the shards and their messages
Two-phase commit turns many shards into one atomic decision. In the prepare phase the coordinator asks each shard whether it can commit; a shard logs its change durably and votes yes, or votes no. The coordinator commits only if every vote is yes, logs that decision, then in the commit phase tells all shards to apply it. One no anywhere forces a global abort, so the transaction is all-or-nothing across machines.
Notes ↗
M5

Consistency in Replicas

Replicas keep copies of the data on several machines, so it survives a dead disk or a dead node.

4 figures
consensus-leader-election · Option 1: Write to All Replicas

Option 1, write to all replicas: every copy must confirm, so one slow or dead replica stalls the write

Option 1, write to all replicas: every copy must confirm, so one slow or dead replica stalls the write A client sends one write to all three replicas at once. Replica one and replica two acknowledge quickly (green checks). Replica three is slow or crashed (amber), so it never acknowledges. Because the write is only safe once every copy confirms, the client is blocked and cannot reply OK. The lesson: writing to all replicas has no fault tolerance, a single slow or dead copy stalls every write. Color key: green is a replica that acknowledged, amber is the slow or crashed replica that blocks the write, grey is the client and the write it sent. Option 1 — write to all replicas Every copy must confirm, so one slow or dead replica stalls the write. Client write x = 5 Replica 1 Replica 2 Replica 3 write to all three acked acked ! slow / crashed blocked, no OK waits for all three All-or-nothing: one slow or dead replica blocks every write. No fault tolerance. Green: a replica that acknowledged · amber: the slow or crashed replica that blocks the write · grey: the client and its write.
Writing to all replicas has no fault tolerance. A write is safe only once every copy acknowledges, so a single slow or crashed replica (amber) stalls every write while the client waits. In a cluster where machines fail routinely, waiting on all of them is unusable.
Notes ↗
consensus-leader-election · Option 2: Write to a Leader

Why a majority works: any two majorities overlap

Why a majority works: any two majorities overlap Five nodes in a row. One majority is any three of them, for example nodes one, two, and three. Any other majority, for example nodes three, four, and five, must share at least one node with the first, here node three. Because each node votes only once per term, that shared node cannot back two winners, so at most one candidate ever reaches a majority and split-brain is impossible. Color key: pink marks a majority set and the shared node, grey marks the nodes. Why a majority works In a 5-node cluster, any two majorities must overlap. N1 N2 N3 N4 N5 Majority A · any 3 of 5 Majority B · any other 3 N3 is in both. It votes once per term, so only one side ever reaches a majority. No split-brain.
Any two majorities of a five-node cluster must overlap on at least one node. Since each node casts one vote per term, that shared node cannot put two candidates over the top, so at most one leader is ever elected. That same overlap lets a majority both tolerate failures and prevent split-brain.
Notes ↗
consensus-leader-election · Option 2: Write to a Leader

Leader election: one leader fails, the cluster elects another

Leader election: one leader fails, the cluster elects another A three-node cluster over time. In steady state, the pink leader serves clients while two grey followers replicate the log. When the leader fails, it goes grey with a cross and heartbeats stop. A follower whose timeout elapsed becomes a candidate and asks the others for votes. Winning a majority, it turns pink and resumes heartbeats, about 200 ms after the failure. The election only runs when the leader fails; it is not a pipeline of stages. Color key: pink is the leader and the election in focus, grey is the followers and the timeline. Leader election One leader serves. When it fails, the survivors run a quick election. Steady state L Leader serves; followers replicate T = 0 Leader fails No heartbeat; followers wait crash Election C Timed-out follower asks for votes ~150 ms timeout New leader L Wins a majority; resumes heartbeats ~200 ms elected Color key  pink = the leader and the election in focus  ·  grey = the followers and the timeline
Raft elects a new leader after the old one fails. Each node votes at most once per term, so only one candidate can clear a majority, which rules out split-brain. The timeout that triggers the election is randomized per node, so followers rarely become candidates at the same instant. Production Raft systems like etcd and Consul elect a new leader within about 200 ms of the failure.
Notes ↗
consensus-leader-election · Option 2: Write to a Leader

Synchronous vs asynchronous replication

Synchronous vs asynchronous replication Two panels, each with a client, a primary, and a replica. Left, synchronous: the primary forwards the write to the replica and waits for the replica's acknowledgement before it replies OK to the client, so no committed write is lost on failover, at the cost of higher latency. Right, asynchronous: the primary replies OK to the client immediately and ships the write to the replica later, so latency is low but the un-shipped tail can be lost if the primary fails. Color key: pink is the acknowledgement the client waits for, grey is the nodes and messages, a dashed line is replication that can lag. Synchronous vs asynchronous replication Wait for the replica (safe, slower) or reply now (fast, can lose the tail). Synchronous Wait for the replica before replying Client Primary Replica 1 write 2 copy 3 ack 4 OK No committed write lost on failover Higher write latency (full round trip) Asynchronous Reply now, ship to the replica later Client Primary Replica 1 write 2 OK now 3 later (lag) Low write latency (no waiting) Un-shipped tail can be lost on failover Color key  pink = the OK the client waits for  ·  grey = nodes & messages  ·  dashed = replication that can lag
The two modes trade latency against durability. Synchronous replication forwards each write to a follower and waits for its acknowledgement before replying OK, so a failover to that follower loses no committed write, at the cost of a full round trip per write. Asynchronous replication replies OK as soon as the leader has the write and ships it to followers in the background, which is fast but means a leader crash can lose any writes that had not yet shipped. The gap between leader and follower is replication lag.
Notes ↗
M5

CAP Theorem

CAP says a distributed database can guarantee at most two of Consistency, Availability, and Partition tolerance.

1 figure
cap-theorem · Pick Two

CAP: partition tolerance is a given, all three at once is impossible, so your one choice is the consistent side or the available side

CAP: partition tolerance is a given, all three at once is impossible, so your one choice is the consistent side or the available side Three overlapping circles label Consistency, Availability, and Partition tolerance. Partition tolerance is mandatory because real networks partition, so its circle is filled green, a given. The center where all three meet is filled red: having all three at once is impossible. That leaves one live choice, and its two options are colored so you can name them: the purple region, CP, stays consistent and may reject requests during a partition; the blue region, AP, stays available and may serve stale data. Color key: green is partition tolerance, the mandatory given; red is the impossible all-three center; purple is the consistent choice, CP; blue is the available choice, AP. The CAP theorem Partition tolerance is a given. All three at once is impossible. So you choose: consistent or available. Consistency (C) Availability (A) Partition tolerance (P) Partition tolerance (P) mandatory: the network splits all three impossible CP · consistent may reject writes AP · available may serve stale green = P, the given  ·  red = all-three, impossible purple = CP, the consistent choice  ·  blue = AP, the available choice
Partition tolerance is a given, not a choice. Consistency means every read sees the latest committed write. Availability means every request gets a response even when some nodes are unreachable. Partition tolerance means the system keeps operating when the network drops messages between groups of nodes. Real networks fail, so partition tolerance is mandatory (the green circle you always live inside), and during a partition your one choice is C or A: stay consistent and reject what you cannot verify (CP), or stay available and risk serving stale data until it reconciles (AP).
Notes ↗
M5

Distributed Sort & Hash

Distributed sort and hash are the two foundational data-movement primitives in a cluster: *hash partitioning* routes every row with the same key to the same machine (so joins and group-bys can run locally), and *distributed sort*

3 figures
distributed-sort-hash · Distributed Sort: One Total Order Across Machines

Distributed sort in three phases

Distributed sort in three phases Three machines sort a dataset into one global order in three phases. Phase one: each machine sorts its own partition locally. Phase two: the cluster samples the data to pick two split points that divide the value range into thirds. Phase three: machines redistribute rows by split point, so node one holds the lowest range, node two the middle, node three the highest, a total order across the cluster. Color key: pink marks the active phase, grey the machines and data. Distributed sort: three phases Sort local, sample split points, redistribute into one global order. 1 · Sort locally each node sorts its own partition Node 1  1 · 4 · 7 Node 2  3 · 6 · 9 Node 3  2 · 5 · 8 2 · Sample split points pick boundaries that cut the range into equal thirds ≤ 3 4 – 6 ≥ 7 3 · Redistribute by range rows move to the node that owns their range Node 1  1 · 2 · 3 Node 2  4 · 5 · 6 Node 3  7 · 8 · 9 Read the nodes in order: one total sort across the cluster.
Distributed sort in three phases. Each node first sorts its own partition locally. The cluster then samples the data to pick split points that divide the value range into equal pieces (here ≤3, 4 to 6, ≥7). Finally rows redistribute so each node owns one contiguous range, and reading the nodes in order gives a single total sort. Sampling is what keeps each node's share balanced even when the data is skewed.
Notes ↗
distributed-sort-hash · Hash Partitioning: Foundation of Distribution

Distributed hash join: hash both tables on the join key user_id, so every row for a user_id meets on one machine, then join locally

Distributed hash join: hash both tables on the join key user_id, so every row for a user_id meets on one machine, then join locally Three stages left to right. First, two tables joined on user_id: Listens (user_id, song_id, rating) and Users (user_id, name, country), with rows colored by the user_id value (7 blue, 42 orange, 88 purple). Second, both tables are hashed on user_id, so every row with a given user_id, from either table, is routed to the same machine: user_id 7 to machine one, 42 to machine two, 88 to machine three. Third, each machine joins only its own slice on user_id and the partial results are unioned into the full join. No machine compares its rows against another machine's, so throughput scales with the cluster. Color key: the color is the user_id value (the join key); grey arrows route each row by its hashed user_id. Distributed hash join Hash both tables on the join key, and every row for one key meets on one machine 1 · Two tables, join on user_id Listens user_id · song_id · rating 7 · s_19 · 5 42 · s_04 · 4 7 · s_51 · 3 88 · s_23 · 5 Users user_id · name · country 7 · Mickey · US 42 · Minnie · UK 88 · Daffy · US hash(user_id) 2 · Same user_id, same machine Machine 1 · user_id 7 7 · s_19 · 5 · 7 · s_51 · 3 7 · Mickey · US Machine 2 · user_id 42 42 · s_04 · 4 42 · Minnie · UK Machine 3 · user_id 88 88 · s_23 · 5 88 · Daffy · US join locally 3 · Join each slice on user_id, union Each machine joins its own rows Mickey · s_19 · 5, s_51 · 3 Minnie · s_04 · 4 Daffy · s_23 · 5 Union the parts = full join Linear speedup no coordinator, no cross-machine join Color key: the color is the user_id value (the join key) — 7 blue · 42 orange · 88 purple. Grey arrows route each row by its hashed user_id.
A distributed hash join. Both tables are hashed on the join key user_id, so every row for one user_id, from either table, is routed to the same machine (the colour is the key value: 7 blue, 42 orange, 88 purple). Each machine joins only the rows it holds, then the partial results union into the full join. No machine compares its rows against another machine's, so throughput scales with the cluster.
Notes ↗
distributed-sort-hash · Consistent Hashing

Consistent hashing: each key goes to the first node clockwise, and adding a node moves only the small arc behind it

Consistent hashing: each key goes to the first node clockwise, and adding a node moves only the small arc behind it A ring spans 0 to 2^32 minus 1. Three grey nodes A at the top, B lower right, C lower left, sit at hashed positions. Three grey keys sit on the ring, and from each a pink arrow walks clockwise to the first node it meets. Then a new node D is added halfway between C and A. Only the keys in the short green arc between C and D move: key 1 leaves A and goes to D. Key 2, which sits past D between D and A, still walks clockwise to A and does not move; nor does key 3. So adding a node shifts only that one small arc, about 1 over N of the keys, not the whole side. Color key: pink is clockwise ownership; green is the small arc of keys that moves to the new node; grey is the ring, the nodes, and the keys. The consistent hashing ring Each key belongs to the first node clockwise. Add a node, and only its short arc of keys moves. A Node A B Node B C Node C key 1 key 2 key 3 D Node D (new) only C to D moves key 1 leaves A for D key 2 is past D: still A, does not move Adding a node moves only the small arc behind it, about 1/N of the keys, not the ~75% modulo reshuffles. Color key  pink = clockwise ownership  ·  green = the short arc that moves to the new node  ·  grey = ring, nodes, keys
Consistent hashing places both keys and machines on a ring over 0 to 2³²−1; each key belongs to the first node clockwise from it. A new node steals only the keys between it and its predecessor, about 1/N of them, so the cluster grows without the near-total reshuffle hash % N forces. DynamoDB, Cassandra, memcached pools, and CDN load balancers all run on this.
Notes ↗
M5

Case Study: How did Google store the entire internet on cheap, dying hard drives?

4 figures
case-study-gfs · How did Google store the entire internet on cheap, dying hard drives?

Cheap hardware at scale: failure becomes routine

Cheap hardware at scale: failure becomes routine Google chose commodity disks over a few expensive enterprise drives. At ten thousand cheap disks, a failure every day is a certainty, so the software has to treat failure as normal. The figure shows the choice (commodity chosen over enterprise), the scale (a grid of ten thousand disks), and the reality (some disks fail every day). Color key: pink is the commodity choice and its consequence, red is a failed disk, grey is the working disks. Cheap Hardware, at Scale Google linked thousands of commodity disks instead of a few costly ones. Enterprise drives reliable, but cost-prohibitive ✗ vs Commodity disks cheap — chosen 10,000 commodity disks A disk fails every day — by design. So the software treats failure as routine, not an emergency. Color key  pink = the commodity choice & its consequence  ·  red = a failed disk  ·  grey = working disks
Cheap hardware at scale. Google linked thousands of commodity disks instead of a few costly enterprise drives. Across ten thousand disks a failure every day is a certainty, shown by the red disks in the grid, so the software treats disk loss as routine.
Notes ↗
case-study-gfs · How did Google store the entire internet on cheap, dying hard drives?

Sharding: capacity beyond one disk

Sharding: capacity beyond one disk A 100 terabyte file cannot fit on a 1 terabyte disk, so GFS splits it into 64 megabyte chunks and scatters them across many machines. The figure shows the mismatch (file far larger than a disk), the split into millions of fixed-size chunks, and the chunks spread across machines so total capacity grows with the cluster. Color key: pink is the chunks and the capacity that scales, grey is the file, disk, and machines. Sharding: Capacity Beyond One Disk Split the file into 64 MB chunks and scatter them across machines. 100 TB web crawl (one logical file) 1 TB disk ✗ does not fit Split into 64 MB chunks — about 1.6 million of them … Machine 1Machine 2Machine 3… Total capacity grows with every machine you add. Color key  pink = the 64 MB chunks & the capacity that scales  ·  grey = the file, disk & machines
Sharding for capacity. A 100 TB file cannot fit on a 1 TB disk, so GFS splits it into 64 MB chunks, about 1.6 million of them, and scatters them across machines. Total capacity then grows with every machine added to the cluster.
Notes ↗
case-study-gfs · How did Google store the entire internet on cheap, dying hard drives?

GFS: a leader directory over cheap chunk servers

GFS: a leader directory over cheap chunk servers A client, a leader, and three chunk servers. The leader keeps an in-memory directory mapping each file to its ordered chunks and each chunk to the servers holding a replica. To read, the client first asks the leader for a chunk's replica locations, the leader replies with the chunk handle and server list, then the client reads the chunk directly from a chunk server. Every chunk is stored as three replicas on different servers, shown by chunk C7 appearing on all three. Color key: pink is the leader's reply and the replicated chunk, grey is the nodes and messages. Google File System One leader directs reads; cheap chunk servers hold replicated 64 MB chunks. Client Leader in-memory directory file → chunk list · chunk → replica servers Chunk Server 1 Chunk Server 2 Chunk Server 3 C3 C9 C1 1 · (file, byte offset) 2 · chunk handle + replica servers 3 · read directly C7 C7 C7 Every chunk = 3 replicas on different servers (chunk C7 shown on all three) Color key  pink = the leader's reply & the replicated chunk  ·  grey = nodes & messages
The GFS lookup. Data never streams through the Leader: the client asks it only for a chunk's replica locations (steps 1 and 2), then reads the chunk directly from a Chunk Server (step 3). The Leader holds just the directory, small enough to keep in memory. Every chunk is stored as three replicas on different servers, here chunk C7 on all three, so a dead disk costs nothing.
Notes ↗
case-study-gfs · How did Google store the entire internet on cheap, dying hard drives?

Replication: self-healing storage

Replication: self-healing storage Every 64 megabyte chunk is stored as three identical replicas on different chunk servers, ideally on separate racks. When one server dies, two replicas still serve, and the leader copies the chunk to a fresh healthy server to restore three copies. The figure shows the three replicas, a server failing with two surviving, and the leader re-replicating to a new server. Color key: pink is the replicated chunk and the recovery copy, red is the dead server, grey is the servers and leader. Replication: Self-Healing Storage Three copies of every chunk; a dead server triggers a fresh one. Leader tracks each chunk's replica servers Every chunk is stored as 3 replicas on different racks Chunk Server 1Chunk Server 2Chunk Server 3 Rack 1Rack 2Rack 3 C7C7C7 server down A server dies — two replicas still serve. Chunk Server 4 healthy, fresh C7 copy C7 The Leader copies C7 to a healthy server — back to three. Color key  pink = the replicated chunk & the recovery copy  ·  red = the dead server  ·  grey = servers & leader
Replication and self-healing. Every chunk is stored as three identical replicas on different racks. When a server dies, two replicas keep serving, and the Leader copies the chunk to a fresh healthy server to restore three copies.
Notes ↗
M5

Distributed File Systems

A distributed file system splits each large file into fixed-size chunks, replicates each chunk across many cheap commodity servers, and gives clients a single-namespace view, so the cluster delivers sequential read throughput prop

2 figures
distributed-file-systems · The Pattern, and What Changes Across Generations

Distributed file systems: how metadata coordination evolved

Distributed file systems: how metadata coordination evolved Four columns, left to right by year. GFS (2003) and HDFS (2006) put all metadata on one pink node, a single Master or NameNode, which caps them near petabyte scale (64 MB / 128 MB chunks, 3x replication, CP). S3 (2006) spreads metadata across many pink nodes with no leader and scales without limit (redundant across at least 3 availability zones, designed for 11 nines of durability; strongly read-after-write consistent since 2020, earlier eventually consistent). Colossus (2010+) shards metadata across many nodes with Reed-Solomon erasure coding and reaches exabyte scale (CP, about 50 percent storage overhead). The grey boxes are the chunk or block servers underneath. Color key: pink is the metadata-coordination tier; grey is the chunk servers underneath. Distributed file systems Same chunks-and-replicas substrate; metadata coordination is what evolved. 2003 GFS Master all metadata CS CS CS 64 MB chunks · 3× CP · ~PB scale Master = bottleneck 2006 HDFS NameNode all metadata DN DN DN 128 MB · 3× rack-aware CP · ~PB scale same bottleneck 2006 S3 md md md no leader AZ AZ AZ variable · ≥3 AZs (erasure) leaderless · ∞ scale strong RAW (since 2020) 2010+ Colossus md md md distributed metadata RS RS RS variable · Reed-Solomon CP · EB+ scale ~50% overhead Color key  pink = the metadata-coordination tier  ·  grey = the chunk servers underneath
Four generations share one chunks-and-replicas substrate but coordinate metadata differently, and that single choice sets each scale ceiling. GFS (2003) and HDFS (2006) keep all metadata on one Master or NameNode (64 MB / 128 MB chunks, 3× replication, CP), so the single leader caps them near petabyte scale. S3 (2006) drops the leader for distributed metadata (redundant across ≥3 availability zones, designed for 11 nines of durability) and scales effectively without limit; since 2020 it is strongly read-after-write consistent (earlier, eventually consistent). Colossus (2010+) shards metadata across many nodes with Reed-Solomon erasure coding (~50% overhead versus 200% for 3× replication, CP) and reaches exabyte scale.
Notes ↗
distributed-file-systems · Colossus (2010+): GFS Without the Bottleneck

Colossus: Reed-Solomon erasure coding

Colossus: Reed-Solomon erasure coding Instead of keeping three full copies, Colossus splits data into six data blocks and computes three parity blocks from them, nine blocks in all spread across machines. Any three blocks can be lost and the original data is rebuilt from the six that survive. This tolerates three failures at 50 percent storage overhead, versus three-way replication, which tolerates two failures at 200 percent overhead. Color key: pink is the parity blocks and the erasure-coding win, red is a lost block, grey is the data blocks. Colossus: Reed-Solomon Erasure Coding Survive 3 failures at 50% overhead, not the 200% of three copies. D1D2D3D4D5D6 P1P2P3 6 data blocks 3 parity (computed from the data) 6 data + 3 parity = 9 blocks, spread across machines Lose any 3 of the 9 blocks The math rebuilds them from any 6 survivors. Reed-Solomon (6, 3): +50% storage · survives 3 failures Three copies: +200% storage · survives 2 failures Color key  pink = parity & the erasure-coding win  ·  red = a lost block  ·  grey = data blocks
Reed-Solomon erasure coding. Colossus splits data into data blocks and computes parity blocks from them, for example six data plus three parity. Any three of the nine can be lost and rebuilt from the six that survive, so it tolerates three failures at 50% storage overhead, where three full copies cost 200% and survive only two failures.
Notes ↗
M5

MapReduce: Divide and Conquer at Scale

MapReduce is a programming model where you write two functions, map(record) → (key, value) and reduce(key, listvalue]) → result, and the framework distributes the work across thousands of machines using hash partitioning to shuffl

1 figure
mapreduce · A Worked Example: Top Songs Across 100B Plays

MapReduce: top songs across 100 billion plays

MapReduce: top songs across 100 billion plays Left to right: 100 billion log lines in HDFS feed 1,000 grey mappers that each emit (song_id, 1). The pink shuffle hash-partitions pairs by hash(song_id) % R so the same key always reaches the same reducer; this stage is network-bound and writes to disk. 100 grey reducers each sum the 1s for their slice of song IDs, producing about 50 million per-song counts. Color key: pink is the shuffle, the hidden core; grey is the input, mappers, reducers, and output. MapReduce: top songs across 100B plays Map emits pairs, the shuffle routes them by key, reduce sums per key. INPUT Chunk 1 ~100M lines Chunk 2 ~100M lines Chunk 3 ~100M lines 100B lines in HDFS MAP Mapper 1 emit(song_id, 1) Mapper 2 emit(song_id, 1) Mapper 3 emit(song_id, 1) 1,000 mappers SHUFFLE hash(song_id) % R routes pairs by key network-bound, to disk REDUCE Reducer A ~500K songs Reducer B ~500K songs 100 reducers OUTPUT (song_42, 9.1M) (song_99, 8.4M) (song_27M, 7.7M) … ~50M counts Color key  pink = the shuffle, the hidden core  ·  grey = the input, mappers, reducers, and output
MapReduce finds the top songs across 100 billion plays. 1,000 mappers each read one 3×-replicated chunk and emit (song_id, 1) per play. The shuffle partitions pairs by hash(song_id) % R so the same key always reaches the same reducer. Each reducer sums the 1s for its slice of song IDs and writes the per-song counts. The same code runs in about 10 minutes on 1,000 machines instead of weeks on one.
Notes ↗
M5

Apache Spark: MapReduce Evolved

Spark generalizes MapReduce by representing a dataset as a *Resilient Distributed Dataset* (RDD): a partitioned, lazily-evaluated, lineage-tracked collection.

7 figures
spark · RDDs: The In-Memory Pipeline

An RDD is a partial result

An RDD is a partial result Four boxes left to right, each holding a small table, joined by labelled arrows. The first box is RDD zero, the Listens table with nine rows, the data as it arrives. An arrow labelled filter leads to RDD one, the same rows with the non-pop ones gone. An arrow labelled group leads to RDD two, which holds one row per user: Mickey, Minnie and Daffy, with a note that Pluto has no listens and so never appears. A final arrow labelled sort leads to RDD three, the same three users with most plays first. A closing line says each box is the data as it stands after one step, that Spark keeps them in memory rather than writing them to disk, and that the last one is the only box anybody asked for. An RDD is a partial result One query, four steps. Every step leaves behind the data as it stands so far. RDD0 Listens 9 rows, as they arrive What you started with. filter RDD1 only pop listens the rest are gone Fewer rows, same shape. group RDD2 one row per user Mickey Minnie Daffy Pluto has no listens A different shape entirely. sort RDD3 most plays first Minnie Mickey Daffy the answer The only box you asked for. SCAFFOLDING MapReduce wrote each of these three to disk before the next step could read it. Spark keeps them in memory, which is the whole speedup.
One query, four RDDs. Every step leaves behind the data as it stands so far, and only the last box is the answer anybody asked for. The other three are working state. MapReduce wrote each of them to disk so the next step could read it back; Spark keeps them in memory. Pluto never appears after the grouping, because Pluto has no listens.
Notes ↗
spark · RDDs: The In-Memory Pipeline

Same multi-stage pipeline: MapReduce writes to disk, Spark stays in RAM

Same multi-stage pipeline: MapReduce writes to disk, Spark stays in RAM Two panels, same three-stage analytics pipeline (filter to pop, count by user, sort the top). Left, MapReduce runs it as three chained jobs and materializes each job's output to HDFS before the next reads it, about 3 minutes for three disk writes. Right, Spark chains the same three stages into one lazy pipeline that an action triggers and keeps the intermediate RDDs in RAM between stages, about 2 seconds with zero disk writes. The pink elements are the staging medium that sets the speed: the HDFS writes on the left, the RAM pipeline on the right. Color key: pink is the staging medium that sets the speed, grey is the shared operators and structure. Same 3-stage pipeline: MapReduce vs Spark MapReduce writes to HDFS between every stage; Spark keeps RDDs in memory. MapReduce · 3 jobs, 3 disk writes Job 1: Filter genre = pop HDFS Job 2: Group + count by user HDFS Job 3: Sort top by count HDFS Total time ~3 min 3 disk writes Spark · 1 pipeline, 0 disk writes textFile() filter() reduceByKey() sortBy() RDDs flow through RAM no stage hits disk collect() collect() is the action that triggers the lazy pipeline Total time ~2 sec RDDs in RAM Color key  pink = the staging medium that sets the speed  ·  grey = the shared operators and structure
The same three-stage analytics pipeline (filter to pop, count by user, sort the top), two engines. MapReduce runs it as three chained jobs and materializes each job's output to HDFS before the next reads it (about 3 minutes); Spark chains the same three stages but keeps intermediate RDDs in memory and evaluates lazily, building a plan it runs once when an action triggers it (about 2 seconds). The ~100× gap comes from data staying in RAM between stages, operator fusion from lazy evaluation, and cheap lineage-based recovery instead of replication.
Notes ↗
spark · Resilience: Lineage, Not Replicas

Spark recovers a lost partition by replaying its lineage

Spark recovers a lost partition by replaying its lineage Four RDD stages left to right: RDD0 input, RDD1 = filter(), RDD2 = map(), RDD3 = map(), each with three grey partitions P0 to P2. These are all narrow transformations, so each output partition depends on exactly one input partition. The worker holding RDD3.P1 dies, so that partition is marked in pink as lost. Spark reads the lineage and re-runs filter then map then map on just the still-alive upstream RDD0.P1, the pink replay path, with no replicas stored. Color key: pink is the lost partition and its replay path; grey is the intact lineage. Lineage recovery: replay, do not replicate Spark stores the operator graph, not copies of the intermediate data. RDD0 (input) P0 P1 P2 RDD1 = filter() P0 P1 P2 RDD2 = map() P0 P1 P2 RDD3 = map() P0 P1 ✗ node died P2 Replay filter() → map() → map() on RDD0.P1 only Recovery  Spark reads the lineage: RDD3.P1 = map(map(filter(RDD0.P1))) Color key  pink = the lost partition and its replay path  ·  grey = the intact lineage
Spark recovers lost data by replaying lineage instead of keeping replicas. Each RDD records the transformation that produced it (filter, map, map), so Spark stores the operator graph, not the intermediate data. These are all narrow transformations, so each output partition depends on exactly one upstream partition: when a worker dies and takes RDD3.P1 with it, Spark re-runs filter → map → map on just the still-alive RDD0.P1. Storage drops from 3× to 1×, and the cost of a failure rises from reading a replica to recomputing one lost partition. A wide transformation like groupBy would instead re-read every upstream partition through a shuffle.
Notes ↗
spark · Spark SQL: DataFrames + Catalyst

Spark SQL: the same SQL text, executed distributed

Spark SQL: the same SQL text, executed distributed The top box holds the Spotify genre-by-week analytics query, a join across listens and songs, identical text to single-node SQL. The lower-left grey box lists Catalyst's execution plan: broadcast the smaller songs table to every node, join it locally against each listens partition, group by genre and week, count distinct users. The lower-right pink box is the payoff: PostgreSQL on one machine takes about 2 hours for 1 TB, spilling to disk on a single node, while Spark on 100 machines finishes in about 3 minutes and scales linearly. Color key: pink is the distributed-scale payoff; grey is the query and the execution plan. Spark SQL: same syntax, distributed scale Catalyst compiles the same SQL text into distributed RDD operations. Spotify analytics query (same text as the OLAP section) SELECT s.genre, DATE_TRUNC('week', l.listen_time) AS week, COUNT(DISTINCT l.user_id) AS unique_listeners FROM listens l JOIN songs s ON l.song_id = s.song_id Catalyst execution plan 1. Broadcast the small songs table to all nodes 2. Join locally on each listens partition 3. Group by genre and week 4. Count distinct users per group Same query, two systems PostgreSQL (one machine): ~2 hours for 1 TB · disk-bound on one node Spark (100 machines): ~3 minutes for 1 TB · scales linearly Color key  pink = the distributed-scale payoff  ·  grey = the query and the execution plan
Spark SQL runs the same SQL as Postgres but executes it distributed. The Spotify genre-by-week analytics query (a join across listens and songs) is identical text; Catalyst compiles it to RDD operations: broadcast the smaller songs table to every node, join it locally against each listens partition, then group and count. PostgreSQL on one machine takes hours at 1 TB, spilling the join and sort to disk on a single node with no cluster to parallelize across, while Spark on 100 machines finishes in minutes and scales linearly with cluster size. No code changes between single-node and distributed: the optimizer picks join strategies and partition pruning the same way a single-node optimizer does.
Notes ↗
spark · Spark at Real-World Scale: Netflix Recommendations

One query, two surfaces

One query, two surfaces The same query written twice, side by side on two dark code cards. On the left it is SQL: select r dot user id and the average of r dot rating as avg rating, from user ratings r, join movie features f on r dot movie id equals f dot movie id, group by r dot user id. On the right it is the DataFrame chain: user ratings dot join movie features on movie id, dot groupBy user id, dot agg avg of rating aliased to avg rating. Four highlights walk the two panes together, matching the table, the join, the grouping and the average across both languages. The highlights land on different lines in each pane, because SQL is written in result order while the chain is written in execution order. A closing panel notes that Catalyst compiles both to the same plan. One query, two surfaces The same join, group and average. Write it either way. SQL DATAFRAME CHAIN SELECT r.user_id, AVG(r.rating) AS avg_rating FROM user_ratings r JOIN movie_features f ON r.movie_id = f.movie_id GROUP BY r.user_id (user_ratings .join(movie_features, "movie_id") .groupBy("user_id") .agg(avg("rating") .alias("avg_rating"))) FROM user_ratings r (user_ratings The table. A terabyte of ratings, the same starting point on both sides. JOIN movie_features f ON r.movie_id = f.movie_id .join(movie_features, "movie_id") The join. Matched on movie_id, spelled out in full on the left. GROUP BY r.user_id .groupBy("user_id") The grouping. One row per user, on both sides. AVG(r.rating) AS avg_rating .agg(avg("rating") .alias("avg_rating"))) The average. Look where it sits: near the top in SQL, last in the chain. Catalyst compiles both to the same plan
One query, two surfaces. Read the highlights across: the table, the join, the grouping and the average are the same four operations either way, and Catalyst compiles both to the same plan, so you can write whichever one you prefer. The one real difference is order. SQL puts the average near the top, because it states the result you want; the chain puts it last, because it states the steps in the order they run. Both are lazy, so an action such as show() asks for the rows.
Notes ↗
spark · When the Map Step Is a Model Call

The map slot takes a model

The map slot takes a model A Songs table split into four partitions feeds one map slot. First the slot holds an ordinary deterministic function, upper of title, and each partition returns a value computed locally. Then the same slot holds a model call, ai_classify over the title, and each partition instead sends its rows to a model endpoint, drawn in pink. The executor lane underneath is identical in both cases: the same four executors, the same retries, the same lazy plan. A closing line notes that partitioning, the DAG and recovery are unchanged, and only the function in the slot changed. The map slot takes a model Same partitions, same executors, same plan. Only the function changed. SONGS, PARTITIONED P1  25M rows P2  25M rows P3  25M rows P4  25M rows THE MAP SLOT upper(title) a plain function computed on the worker itself THE SAME SLOT ai_classify(title, ...) model endpoint one call per row THE EXECUTOR LANE, UNCHANGED executor 1 P1 executor 2 P2 executor 3 P3 executor 4 P4 The driver scales how many calls are in flight; retries, stragglers and the lazy plan are the ones you already know. Partitioning, the DAG and recovery stay exactly as they were. One thing does change. The next figure shows it.
The slot that held upper(title) now holds a model endpoint, and the engine underneath is unchanged: the same four partitions, the same executors, the same lazy plan and the same retries. Only the function moved off the worker and onto an endpoint. At this scale the hard parts are the ones this module already covers: partitioning, retries, stragglers and cost. The model is the easy part.
Notes ↗
spark · What Breaks: Lineage Meets Non-Determinism

Replay gets a different answer

Replay gets a different answer Two tracks over the same lineage. On the top track a deterministic function, upper of title, produces the value ROCK; the worker holding it dies, Spark replays the lineage, and the value comes back identical, marked with a green tick. On the bottom track the same slot holds a model call, ai_classify, which first produced the label workout; the worker dies, Spark replays, and the model returns party instead, marked with a red cross, so the recovered partition disagrees with the one it replaced. A closing panel lists the three fixes used in practice: pin the sampling to temperature zero, cache the answer against a hash of the input row, and materialise inference output to a table, which fixes the labels for good. Replay gets a different answer Lineage recovery assumes the same input gives the same output. A model breaks that assumption. DETERMINISTIC FUNCTION RDD0.P1 still alive upper(title) "ROCK" worker dies, Spark replays ✓ same value, exactly MODEL CALL IN THE SAME SLOT RDD0.P1 still alive ai_classify(...) "workout" first run worker dies, Spark replays ✗ comes back "party" The recovered partition now disagrees with the one it replaced, and the job still reports success. Two rows with identical input can carry different labels, decided by which machine happened to fail. WHAT PRODUCTION PIPELINES DO ABOUT IT Pin the sampling Temperature zero narrows the spread, though the token can still move between runs. Cache on the input Key the answer by a hash of the row, so a replay reads the first answer instead of asking. Materialise the output Write the labels to a table. Downstream stages read them, so the labels are fixed for good.
The same recovery that makes Spark cheap makes a model pipeline inconsistent. Production pipelines close the gap three ways, and only the last one closes it completely: temperature zero narrows the spread while the token can still move; caching the answer against a hash of the input row makes a replay read rather than ask; and materialising inference output to a table removes recomputation entirely, so real pipelines write predictions down once and read them after that.
Notes ↗
M5

BigQuery: Serverless OLAP at Cloud Scale

BigQuery is a serverless OLAP query engine: you submit SQL, Google's compute layer auto-provisions thousands of ephemeral "slots" against shared columnar storage, and you get an answer in seconds.

2 figures
bigquery · The Architectural Shift: Decoupled Storage and Compute

BigQuery is serverless OLAP: storage and compute scale separately

BigQuery is serverless OLAP: storage and compute scale separately An analyst submits SQL. BigQuery spins up roughly 2,000 ephemeral compute slots, the pink layer, that scan in parallel and then vanish, billed about 6 dollars and 25 cents per tebibyte scanned. They read a 10 TB table held as persistent per-column files in Colossus, the grey storage layer, billed about 20 dollars per terabyte per month at rest. The two layers scale and bill independently. Color key: pink is the ephemeral compute in focus, grey is the persistent storage and the query. BigQuery: serverless OLAP Storage and compute scale and bill independently. Analyst submits one SQL query Compute · ~2,000 ephemeral slots slot slot slot slot slot … ~$6.25/TiB scan return Storage · Colossus, persistent columnar files artist file year file user_id file song_id file … 10 TB total ~$20/TB/mo Color key  pink = the ephemeral compute in focus  ·  grey = the persistent storage and the query
BigQuery is serverless OLAP: storage and compute scale and bill independently. Each query spins up roughly 2,000 ephemeral slots that run in parallel and vanish when it finishes, billed per byte scanned (about $6.25/TiB). Colossus holds the table as persistent columnar files, one column per file, billed per gigabyte at rest (about $20/TB/month). No cluster to provision: you pay for the bytes you scan, not for idle capacity.
Notes ↗
bigquery · A Worked Query: Spotify Top Artists, 100B Plays

Behind a BigQuery API call: five stages, from planning a columnar scan to returning the top 100 in about 30 seconds

Behind a BigQuery API call: five stages, from planning a columnar scan to returning the top 100 in about 30 seconds One query flows through five stages, left to right. 1, Plan: the optimizer reads the query and schema, sees only artist and year are touched, and the columnar layout reads just those two columns and skips the rest, so bytes scanned drop from 10 TB to about 200 GB. 2, Schedule: the slot scheduler allocates up to about 2,000 ephemeral compute slots, split across the stages. 3, Scan in parallel: each slot reads a partition of the columnar files in Colossus and hash-aggregates by artist locally into partial counts. 4, Shuffle and merge: a second wave of slots receives the partial counts, hash-partitioned by artist, sums them, sorts, and trims to the top 100 (the same hash partitioning as Distributed Sort and Hash). 5, Return: the result lands in the client in about 30 seconds, slots release, and you pay only for the bytes scanned, about $1.25. Color key: pink is the key number or result at each stage; grey is the stage boxes and the flow between them. Behind the API call: one query, five stages SELECT artist, COUNT(*) FROM listens_2026 WHERE year=2026 GROUP BY artist ORDER BY plays DESC LIMIT 100 1 · Plan Optimizer reads the query and the table schema. Only artist + year are read; columnar skips the rest, and those two compress hard. 10 TB → 200 GB bytes scanned 2 · Schedule The slot scheduler hands out ephemeral compute slots, split across the query's stages. ~2,000 slots on-demand default 3 · Scan in parallel Each slot reads one partition of the columnar files in Colossus, and hash-aggregates by artist. partial counts one per slot, local 4 · Shuffle + merge A second wave of slots gets the partial counts, hash-partitioned by artist, sums, sorts, trims. top 100 same hash-partitioning as Sort & Hash 5 · Return The result lands in the client, and the slots release. You pay only for the bytes you scanned. ~30 s · $1.25 nothing idle to pay for Color key: pink is the key number or result at each stage; grey is the stage boxes and the flow between them.
Behind one API call, the query runs as five stages. Plan: the optimizer identifies artist and year as the only referenced columns, so the columnar layout (from the Module 3 BigQuery case study) reads only those two columns and skips the rest; over 100 billion rows they compress to ~200 GB, a fraction of the full 10 TB. Schedule: ~2,000 ephemeral slots are split across the stages. Scan: each slot reads a Colossus partition and hash-aggregates by artist into partial counts. Shuffle and merge: a second wave hash-partitions those counts by artist, sums, sorts, and trims to the top 100, the same hash-partitioning as Distributed Sort & Hash. Return: the result lands in ~30 s, slots release, and you pay only for the ~200 GB scanned, about $1.25.
Notes ↗
M5

Apache Kafka: Real-Time Event Streaming

Kafka is a durable, partitioned, replicated commit log: producers append events to per-topic partitions, consumers read them at their own pace, and the log itself is the source of truth, so multiple downstream systems can independ

1 figure
kafka · A Worked Example: One Play, Two Pipelines

Kafka is a partitioned, replicated commit log

Kafka is a partitioned, replicated commit log The Spotify client produces a play_started event. The producer hashes user_id to pick one of the topic's three pink partitions, P0 to P2, each an ordered append-only log replicated 3x across brokers. Two independent grey consumers read at their own offsets: the recommendation pipeline at offset 16 in real time, the royalty pipeline at offset 12 nightly. Because events persist until retention expires, a consumer can rewind and replay any past offset. Color key: pink is the durable partitioned log; grey is the producer and the independent consumers. Kafka: a partitioned, replicated commit log Events persist, so independent consumers read and replay the same log. Spotify client play_started hash(user_id) Topic: plays · 3 partitions, 3× replicated P0 12 13 14 15 16 new P1 7 8 9 P2 21 22 23 24 offset 16 offset 12 Recommendation pipeline streams in real time Royalty payouts batches nightly Color key  pink = the durable partitioned log  ·  grey = the producer and the independent consumers
Kafka is a partitioned, replicated commit log. The producer hashes each event's key (here user_id) to pick a partition, so one user's events stay in order; the topic's 3 partitions each replicate 3× across brokers, giving parallelism and ordering from partitions and durability from replicas. Two independent consumers, Recommendations and Royalty, track their own offsets and read at their own pace, and because events stay until retention expires, either can rewind and re-read. Typical scale: millions of events per second, under 10 ms end to end, retained for days or longer.
Notes ↗
end of lecture 1~54 min of figures
M5 Distributed systems 1 of 29
Notes ↗ Video ▶ Why ◎ Schedule ↗ □ keys

M5 Distributed systems

Sharding, consensus, GFS, MapReduce, Spark · 29 figures · ~54 min · 1 lectures at this pace