Unit 2 · 5 hrs · CSUE301 Big Data Analytics

Distributed
Storage Systems

Chapter 1 asked what makes data big and how a cluster processes it. This chapter asks the question underneath that one: once a byte is written, where does it actually live? We stay with CityCourier's sixteen-order log — same rows, same three hubs — but now we open the floor and look at the disks.

518 bytes, three blocks, two copies each

A file is not one thing

It's pieces, and the pieces have copies.

The whole order log is 518 bytes — tiny, but the mechanics don't care about size. It gets cut into blocks by arrival order and each block is stored on two of the three hub servers, so no single machine holds the only copy of anything. Section 2.1 works this out block by block.

Colour contract for this chapter Hub N's server Hub S's server Hub E's server What the storage layer had to decide
2.1 · BLOCKS, REPLICATION & THE NAMENODE

A File Is Not One Thing

CityCourier's order log sits on Hub E's server tonight. Hub E's server also sits in a building that loses power sometimes. What, precisely, protects the log?

In Chapter 1, "store" was one arrow in a seven-stage pipeline — a black box a record disappears into. This section opens that box. A distributed file system takes one logical file and spreads it, in pieces, across many machines, so that no single disk failure can take the file with it. The design nearly every modern system either runs or imitates is Hadoop's HDFS, itself modelled on Google's GFS — and its two central ideas are worth learning by name, because you will meet variants of both again in Chapter 3.

Two kinds of node, two very different jobs

An HDFS cluster has exactly one active NameNode and many DataNodes. The split is strict, and mixing the two up is the single most common misconception about how this works:

who stores what NameNode stores METADATA ONLY — which blocks make up a file, and which DataNodes hold each block. No file bytes ever pass through it. DataNode stores the actual blocks, on its own local disk, and serves them directly to clients.

Every DataNode sends the NameNode a periodic heartbeat ("I'm alive") and a block report ("here is exactly what I'm holding"). Stop hearing heartbeats from a node, and the NameNode has to assume it's gone — which is exactly the scenario this section's worked example walks through.

Splitting a file into blocks

A file is cut into fixed-size blocks the moment it's written — not by content, not by row, just by raw byte position. Real HDFS defaults to a 128 MB block size, chosen for machines storing terabytes; every full block after the first is exactly that size, and only the last block is shorter. Each block is then replicated some number of times (the replication factor) and each copy is placed on a different DataNode, so losing one machine never means losing a block outright.

why not just mirror the whole disk?

Whole-disk mirroring protects against one failure but wastes a second disk doing nothing but waiting. Block-level replication spreads copies across many machines instead, so the same redundancy also spreads read load across more disks, and a rebuild after a failure only has to recreate the missing blocks, not an entire disk image.

Worked Example · Blocking and Placing the Order Log

CityCourier's 16-row order log, written out one row per order, is 518 bytes total — about 32 bytes a row. That's roughly 0.0004% of one real HDFS block, which is exactly why we can lay out every byte on the page and still see the real mechanics.

Cut the file, place the copies, survive a failure
518 B · 3 blocks · replication factor 2
Split the log into 3 blocks by arrival order, replicate each block twice across Hub N, Hub S and Hub E's servers, and check: does the file survive any one server going down?
1 · cut into blocksBlocks are cut by position, not by hub — Block 1 is orders #1–6 (191 B), Block 2 is #7–11 (161 B), Block 3 is #12–16 (166 B). Because orders arrive interleaved across hubs (the sequence runs N,S,N,E,S,N,E,S,…), every block contains rows from all three hubs. A block is a slice of write order, nothing more.
2 · place 2 copies each, round-robinBlock 1 → {Hub N, Hub S}. Block 2 → {Hub S, Hub E}. Block 3 → {Hub E, Hub N}. Every server ends up holding exactly 2 of the 3 blocks — two-thirds of the file each, no server left idle and none overloaded.
3 · fail Hub SHub S's server drops offline mid-rush. Block 1 now has one live copy (at N). Block 2 now has one live copy (at E). Block 3 is untouched — its two copies were at E and N all along.
4 · the NameNode reactsMissing heartbeats from Hub S tell the NameNode it's gone. Checking its metadata, it finds two under-replicated blocks and issues copy orders to the only servers that can take them: Block 1 gets copied onto Hub E (the one live server that lacked it); Block 2 gets copied onto Hub N. Nothing was lost, and nothing about this required a human to notice.
Answer: yes — every block kept at least one live copy through the failure, so the file survived intact. It survived because replication factor 2 was chosen; replication factor 1 would have lost Block 1 and Block 2 outright the moment Hub S went down.
Watch It Happen
Fail a node, step through the NameNode's response
3 blocks · 3 servers · RF=2
Step 1 of 3
All three blocks sit at full replication — two live copies each, spread so that no server holds every copy of anything.
pitfalls
  • "The NameNode stores the file." It stores the map of the file — which blocks, which DataNodes. Every byte of actual content lives only on DataNodes. This is also why classic HDFS treated NameNode failure as so serious: lose the map (with no backup) and the blocks on disk become unreadable, even though every byte is technically still sitting right there.
  • "Replication factor 3 means the file always takes up 3× the space." Replication factor is set per file and can change after creation. A scratch file might run at 1, a critical file at 5. "3" is only Hadoop's default, not a law.
  • "More replicas is free safety." Every replica is a full write, on every write. Replication factor 2 already doubled Section 1.6's "moving data costs bytes" bill for this file; factor 5 would 5× it. Durability and write cost trade directly against each other.
Practice
  1. If CityCourier's log grew to 40 orders and kept the same 3-block, round-robin, replication-factor-2 layout, how many total block-copies would exist on disk, and how many would each server hold?
    Solution

    3 blocks × 2 copies = 6 block-copies total, regardless of how many orders the blocks contain — the block count doesn't change just because the blocks got bigger (until a block exceeds the size limit and a 4th block is created). Each of the 3 servers holds exactly 2 of the 3 blocks, same as with 16 orders.

  2. Suppose the round-robin placement had instead put replication factor 1 everywhere — Block 1 only on N, Block 2 only on S, Block 3 only on E — to save space. Redo the Hub S failure. What's lost?
    Solution

    Block 2 is gone completely — orders #7–11, with no surviving copy anywhere. Blocks 1 and 3 are untouched since they never lived on Hub S. This is exactly the gap replication factor 2 closed in the worked example: at RF=1, a node failure doesn't just risk data, it guarantees loss of whatever that node uniquely held.

  3. After the NameNode's re-replication, Hub N and Hub E each hold all 3 blocks, and Hub S (once it comes back online) holds none of the fresh copies. Is this a problem, and if so, whose job is it to fix?
    Solution

    It's an imbalance, not a data-loss risk — every block is still safely at 2 copies. But it does mean N and E are now doing more than their share of the read load, and a returning Hub S is sitting idle. Real HDFS runs a background balancer process for exactly this: it doesn't fix urgent under-replication (the NameNode's re-replication already did that), it fixes long-run unevenness afterward.

  4. Chapter 1's cost trade-off (Section 1.6) argued that moving computation to data beats moving data to computation when data is large. Does replication factor 2 make that argument stronger or weaker, and why?
    Solution

    Stronger. With RF=2, a computation that needs this file now has two machines it could run on locally instead of one — more chances to find a copy near the requesting client, and more chances to avoid shipping the data anywhere at all. Replication isn't only a durability feature; it's also what makes "run the job where the data already is" a realistic scheduling option rather than a lucky coincidence.

2.2 · NOSQL DATA MODELS

One Order, Four Shapes

Order #4 — Hub E, Sushi Go, ₹560 — is one fact. It can be stored at least four structurally different ways, and the choice changes what's fast, what's flexible, and what's even expressible.

A relational table forces every row into the same fixed columns. NoSQL is the umbrella term for stores that don't — not one alternative design, but four genuinely different ones, each optimised for a different access pattern. Knowing the name of each model matters less than knowing which questions it answers cheaply.

The same order, modelled four ways

order #4 across the four NoSQL models
ModelWhat it looks like hereCheap questionExpensive question
Key–Valueorder:4 → an opaque blob, "Hub E|Sushi Go|130|560|5|900""Give me everything about order 4" (exact key)"Which orders cost over ₹400?" (store can't see inside the blob)
Documenta JSON object with named, nested fields — ratings, timestamps, anything, per document"Give me order 4's amount field only" (structure is visible)joining across many documents efficiently
Column-familyrow key 4, with columns grouped into families (core: hub, restaurant, amount · feedback: rating)writing millions of rows fast; families with no data for a row simply don't exist for itad-hoc queries on a column that isn't the row key
Graphan Order node connected by PLACED_AT to a Hub node and ORDERED_FROM to a Restaurant node"Which orders touch the same restaurant as order 4?" (one hop)"Sum everything" (graphs are bad at bulk aggregation)
the sparse-column callback

Orders #3, #7 and #11 have no rating (Section 1.1's missing-value gap). In a relational table that's a NULL sitting in every one of those rows. In a column-family store, the feedback family for those three rows simply doesn't exist — not a null, an absence. Column-family stores were built for data that is sparse like this at massive scale, where storing an explicit NULL for every missing rating across a billion rows would waste real space.

Worked Example · What Fits in a Byte

The same fact costs a different number of bytes depending which shape holds it — not because any model is "wasteful," but because each is paying for a different kind of flexibility.

order #4, serialized three ways key–value blob 34 B no field names on the wire at all — the application must already know the layout column-family (2 families) 48 B short family/column tags, no nesting document (JSON) 129 B every field name spelled out, every time, for every document

The document format costs nearly 4× the key–value blob for the identical fact — that overhead buys self-describing structure (any tool can read a JSON document without a separate schema) and the ability to query by field name instead of parsing a blob. Whether that's a good trade depends entirely on what you're about to do with the data next.

The Graph View

Order #4's graph neighbourhood looks like this once the other orders that share its restaurant are added for context:

PLACED_AT ORDERED_FROM ORDERED_FROM ORDERED_FROM Order #4 Hub E Sushi Go Order #15 Order #11

Orders #11 and #15 also ordered from Sushi Go — in a graph store, "which other orders share a restaurant with order #4" is one hop away: walk from Order #4 to the Sushi Go node, then out to whichever other Order nodes connect there. A relational answer needs a self-join on restaurant; a graph just walks an edge.

pitfalls
  • "NoSQL means no schema, ever." Document stores are schema-flexible, not schema-free — the application still expects certain fields to usually be there. Column-family stores are stricter still: families are typically defined up front, even if individual columns within them vary row to row.
  • "Pick the trendiest NoSQL model." The four models above answer different questions cheaply. A social network's friend-of-a-friend query is a graph problem practically by definition; forcing it into a column-family store means simulating graph traversal with application code the database isn't helping with at all.
  • "A JSON document is just a slower table row." The 129 B vs 34 B gap above is real cost, but it buys something a fixed table schema can't: two orders in the same collection can have completely different fields, with no schema migration required to add one.
Practice
  1. Order #10 (the veracity bug from Section 1.1) is stored as a document with fields delivery_time_sec: -60. In a column-family store with the same core/feedback split as above, which family holds delivery_time_sec, and does the veracity bug change how it's stored?
    Solution

    It belongs in the core family alongside hub, restaurant and amount — delivery time isn't feedback. And no, the bug doesn't change the storage shape at all: −60 is a perfectly well-typed integer as far as the column-family store is concerned. Veracity is a meaning problem, not a storage problem, which is exactly why catching it requires application-level validation (Section 1.2's Ingest-stage check), not a database constraint.

  2. CityCourier wants to add a "customer allergy notes" field, but only about 2% of orders will ever have one. Which of the four models handles this addition most cheaply, and which would need the most rework?
    Solution

    Document and column-family both handle it cheaply — a document simply omits the field on the other 98%, and a column-family store treats it as another sparse family, exactly like the missing-rating case above. Key–value handles it only if the blob format itself has room to grow, which usually means quietly agreeing on a new blob layout across every piece of code that reads it. A rigid relational table would need an actual schema migration (add a nullable column) across the whole table.

  3. A query asks: "total revenue across all orders." Rank the four models from cheapest to most expensive for this specific question, and justify the graph model's position.
    Solution

    Column-family and document are both cheap here — scan the amount field/column across all rows and sum, exactly Section 1.3's MapReduce pattern. Key–value is more expensive: the store can't see inside the blob, so every value must be fetched and parsed in application code before summing. Graph is worst: summing a property across every node means visiting every node anyway, gaining none of the traversal advantage graphs exist for, while paying extra overhead per node for relationship bookkeeping the query never uses.

  4. Section 1.6 showed that shipping computation to data beats shipping data to computation when data is large. Which of the four models makes that principle easiest to apply for a "sum by hub" query, and why?
    Solution

    Column-family — because hub and amount are both plain columns the store's own engine can filter and aggregate locally, exactly the kind of routine a coordinator can ship to each node and get back only a partial sum (Section 2.6 works this scatter-gather pattern in full). A key–value store can't run that logic itself at all — the "computation" would have to ship out as a separate program that fetches, deserializes, and sums every blob by hand.

2.3 · PARTITIONING & CONSISTENT HASHING

Which Server Owns This Key?

Section 2.1's blocks were sliced by arrival order — the storage layer never looked at what was inside them. A key–value store can't get away with that: given order:4, it has to know exactly which server to ask.

One clarification before anything else: from here on, "Hub N," "Hub S" and "Hub E" mean the server sitting at that hub's location — not the business hub an order was placed at. Those are different questions once you're inside the storage layer, and this section is precisely about how far apart the two answers can drift.

The naive approach, and why it breaks

The obvious scheme is hash(key) mod N: hash the key to a number, take it modulo the number of servers, that's the owner. It works — right up until N changes. Adding or removing even one server changes almost every key's modulo result, because the whole numbering shifts.

Consistent hashing fixes this by hashing servers onto the same numeric space as keys, arranging both on a conceptual ring, and giving each key to whichever server sits next going clockwise. Add or remove a server, and only the keys between the new server's position and its nearest clockwise neighbour move — everyone else's assignment is untouched.

Worked Example · Placing 16 Keys on a Ring

Using a simple illustrative hash — real systems use SHA-1 or Murmur3, but the mechanics are identical — hash(id) = (37×id + 11) mod 100 gives every order id a position from 0–99. Place the three servers at positions N=8, S=42, E=71, and every key belongs to the next server clockwise:

16 keys on the ring, 3 servers
order idhashowner
163Hub N
87Hub N
1118Hub S
322Hub S
1429Hub S
633Hub S
148Hub E
944Hub E
1255Hub E
1566Hub E
770Hub E
459Hub E
1081Hub N
285Hub N
1392Hub N
596Hub N

Sorted by hash, the pattern is visible directly: every key with hash ≤ 8 or > 71 belongs to N (wrapping around the top of the ring), hash 9–42 belongs to S, hash 43–71 belongs to E. Tally it up: N holds 6, E holds 6, S holds 4 — close to even, from just three points on a ring.

the location-transparency payoff

Order #4 was placed at Hub E's counter, and its row happens to hash onto Hub E's own server — a coincidence, not a rule. Order #13, also placed at Hub E, hashes onto Hub N's server instead. Checking all five of Hub E's own orders (#4, #7, #10, #13, #16): two land on Hub E's server, three land on Hub N's, none land on Hub S's. Where a row is stored and which hub it's about are simply unrelated once a hash decides placement — Section 2.6 returns to exactly why that matters for query speed.

Adding a Fourth Server
Add Zone Z and watch the ring redistribute
3 → 4 servers
Three servers, sixteen keys. Every key belongs to whichever server sits next clockwise on the ring.

CityCourier adds a fourth server, Zone Z, at ring position 60 — squarely inside what used to be Hub E's territory (43–71). Only the keys between Hub S's position and Zone Z's new position are affected:

keys reassigned: id 1, 4, 9, 12 (hash 48, 59, 44, 55 — all in the 43–60 gap) old owner: Hub E → new owner: Zone Z everyone else — all 12 remaining keys — keeps the exact same owner moved: 4 of 16 keys (25%)

Compare that to naive hash mod N: moving from 3 servers to 4 changes the modulo result for 11 of the same 16 keys (68.8%) — nearly three times the disruption, for the identical change in cluster size. This is the entire reason consistent hashing exists: the theoretical guarantee is that roughly k/n keys move when a server joins a system already holding k keys across n servers (16/4 = 4, matching exactly), and real systems soften even that with many small virtual nodes per physical server, so no single join or leave dumps its whole share of the disruption onto one neighbour.

pitfalls
  • "Consistent hashing means keys never move." Some do — exactly the ones between the changed server and its predecessor. The property isn't zero movement, it's minimal, bounded movement, which is a very different and achievable promise.
  • "A key's ring position tells you something about its content." It doesn't, by design. Hash functions are built to scatter related keys apart, not cluster them — which is exactly why order #4 and order #13, both Hub E orders, land on different servers.
  • "More servers always means more even load." With only a few ring positions and no virtual nodes, a new server can land unluckily close to an existing one and take almost no load, or land in a wide-open gap and take a lot. Real systems assign each physical server dozens of virtual positions specifically to smooth this out.
Practice
  1. Using the same hash function, order id 17 would hash to (37×17+11) mod 100. Compute it and name the owner, using the original 3-server ring (N=8, S=42, E=71).
    Solution

    37×17+11 = 640; 640 mod 100 = 40. Position 40 falls in the 9–42 range, so Hub S owns it.

  2. If Hub S's server (not a new server, but a failing one) is removed from the original 3-server ring instead of a 4th being added, which keys move, and who inherits them?
    Solution

    Every key that belonged to S (ids 11, 3, 14, 6 — the same four rows that were assigned to S in the worked table) now belongs to whichever server is next clockwise from S's old position, which is Hub E at position 71. No other keys are affected — N's and E's own existing keys are untouched. Removal follows the identical rule as addition: only the range between the removed point and its immediate predecessor changes hands.

  3. A support ticket claims "Hub E's server going down means Hub E's orders are unreadable." Using the ownership breakdown above (2 of Hub E's 5 own orders on Hub E's server, 3 on Hub N's), is the claim accurate?
    Solution

    No — it conflates business hub with storage owner, the exact confusion this section opened by warning against. Hub E's server going down affects whichever keys hash to Hub E's ring position, which happens to include only 2 of Hub E's own 5 orders (plus whichever other hubs' orders also hash there). The other 3 of Hub E's orders live on Hub N's server and are entirely unaffected.

  4. Connect this to Section 2.1: both sections split the same 16-row file across 3 servers, but produce different placements. Why don't blocks and hash-ring ownership agree on which server holds order #4?
    Solution

    They're partitioning at different granularities for different reasons. Section 2.1 blocks the file by raw arrival position for fault-tolerant file storage — it has no notion of "order #4" at all, only byte ranges. This section partitions by an actual application key (the order id) for fast key-based lookup. Nothing requires the two schemes to place the same row on the same server, and in a real system built with both an HDFS-style layer underneath and a key-value layer on top, they generally won't.

2.4 · REPLICATION, QUORUMS & CONSISTENCY

How Many Copies Have to Agree?

Order #4's rating gets corrected from 5 stars to 4. The correction has to reach every replica eventually — but exactly how many replicas have to confirm before the write counts as done, and how many have to be checked before a read counts as trustworthy?

Section 1.6 framed CAP as a single switch: choose Consistency or Availability during a partition. Real systems usually expose something more precise — two dials, called W (write quorum) and R (read quorum), against a fixed replication factor N. A write isn't acknowledged until W replicas confirm it; a read isn't returned until R replicas have responded and their values compared.

the quorum overlap guarantee N total replicas of a key (here, 3 — Hub N, Hub S, Hub E all hold every replicated key) W replicas that must acknowledge a write before it's considered successful R replicas that must respond to a read before it's returned to the caller if W + R > N: every possible read quorum is guaranteed to overlap every possible write quorum

That guarantee isn't a rule someone imposed — it's the pigeonhole principle. If a write touched W of N replicas and a later read checks R of the same N, and W+R exceeds N, there is no way to choose R replicas that avoid all W written ones — there simply aren't enough replicas left over to dodge them all.

Worked Example · Does the Correction Get Seen?

Order #4's rating write goes out to all three replicas, but only two — Hub N and Hub E — acknowledge before a client's write-quorum requirement is met; Hub S is momentarily slow and still holds the old value.

Case A · W=2, R=2 (W+R=4 > N=3)

Every possible 2-of-3 read (there are only three: NS, NE, SE) shares at least one node with the write set {N,E}. All 3 of 3 possible reads see the fresh value — 100%, not a probability, a guarantee.

Case B · W=1, R=1 (W+R=2 ≤ N=3)

The write only had to reach one node — say N. A single-node read might hit N (fresh) or S / E (stale). Only 1 of 3 possible reads sees the fresh value: a 33% chance, and no way to tell which case you're in without checking further.

this is CAP with the dial turned up close

W=R=N (every replica, every time) is maximally consistent but stalls completely the moment any one replica is unreachable — the CP end of Section 1.6's trade-off. W=1, R=1 stays available through almost anything but accepts stale reads — the AP end. N=3, W=2, R=2 (Cassandra and Riak's common default) is the balanced middle: it survives one slow or dead replica in either direction while still guaranteeing overlap. CAP names the trade-off; N/W/R is the actual knob you turn to sit somewhere along it.

pitfalls
  • "W+R>N means the system is strongly consistent, full stop." It guarantees read/write quorum overlap, which is necessary for strong consistency but not sufficient on its own — concurrent writes can still race each other, and which replica's value "wins" when two conflicting writes both reach 2 of 3 nodes still needs a resolution rule (a timestamp, a version vector), not just the quorum arithmetic.
  • "Higher W and R are strictly better." They buy consistency by spending latency and availability: every write now waits on the slowest of W replicas, not the fastest one, and a request fails outright if fewer than the required count are reachable. W=N,R=N stops answering the instant one replica so much as pauses.
  • "N, W and R are one global setting." Real systems typically let W and R be chosen per-operation. CityCourier could write order confirmations with W=1 (fast, availability matters more) and refund reversals with W=3 (slow, correctness matters more) on the very same cluster.
Practice
  1. For N=5, list every (W,R) pair from the set {1,2,3,4,5} where W+R>N, and identify the pair that guarantees overlap while minimising the larger of W and R.
    Solution

    Pairs with W+R>5 include (1,5),(2,4),(3,3),(3,4),(4,2),(4,3),(4,4),(5,1), and every combination with either value at 5. The pair minimising the larger value is (3,3): both below N, both equal, and 3+3=6>5 — the most balanced overlap-guaranteeing choice, which is why odd N with W=R=⌈(N+1)/2⌉ is a common real-world default.

  2. CityCourier switches order confirmations to W=1 for speed, keeping R=2. Does W+R>N still hold at N=3, and what does that imply about reading your own just-written order back immediately?
    Solution

    1+2=3, which is not greater than N=3 — overlap is no longer guaranteed. A customer who writes an order and immediately reads it back with R=2 might hit the two replicas that haven't received the write yet, and briefly see no order at all, even though their own write "succeeded." This is exactly the read-your-own-write problem quorum tuning has to consider explicitly, not assume away.

  3. A read at R=2 returns two different values for the same key from two different replicas. What does the quorum arithmetic alone tell you about which one is correct?
    Solution

    Nothing. W+R>N guarantees you'll see the latest write among the replicas you checked, not that every replica hands back a single agreed value, and it says nothing about how to pick the winner when two exist. That's a separate mechanism — typically a timestamp ("last write wins") or a version vector the application resolves — layered on top of the quorum guarantee, not implied by it.

  4. Section 2.1 showed replication factor 2 surviving a single node failure for file blocks. Does the same N=2 replication factor support a W+R>N quorum with any usable W and R, and what does that reveal about the minimum replication factor a quorum-based design needs?
    Solution

    At N=2, W+R>2 needs W=R=2 (the only pair that works, since 1+1=2 fails and anything above 2 exceeds the replica count). But W=2 at N=2 means every replica must be reachable for every write — a single node failure now blocks all writes outright, the opposite of the fault tolerance Section 2.1 wanted. This is why quorum-based stores typically run N≥3: it's the smallest replication factor that offers a W+R>N choice (like 2-and-2) that still tolerates one replica being briefly unreachable.

2.5 · INDEXING: B-TREE VS LSM-TREE

Optimise the Write, or Optimise the Read

Sixteen orders arrive in a burst. Head office wants to run "show me every order over ₹400" at the end of the day. The index structure that makes writes cheap during the rush is not the one that makes that query cheap afterward.

Both structures below solve the same problem — keep a set of keys (here, order amounts) findable on disk — and both are genuinely in wide production use, because they make opposite bets about which side of the workload to optimise.

two bets on the same problem B-Tree updates IN PLACE, keeping keys sorted on disk at all times. Every insert may shift or split a page — random writes, but reads walk straight to the answer. LSM-Tree never updates in place. Writes land in memory (a "memtable"), and flush to disk as an immutable, append-only file (an "SSTable") once full — sequential writes, but a read may have to check several files.
Worked Example · Inserting the Same 16 Orders, Two Ways

Insert all 16 orders, in arrival order, keyed by amount, into each structure with a small page/memtable capacity of 4 keys, and see what each looks like once the rush is over.

B-Tree: 16 inserts, capacity 4
splits happen when a page overflows
Insert amounts 340,180,260,560,310,420,150,275,355,440,520,165,330,410,545,285 (arrival order) into a sorted B-tree leaf layer. Count the splits.
splits triggered4 splits total — triggered on inserting 310 (5th distinct-page insert), 275, 440, and 410, each time a page's 5th key forced it to divide in two.
final leaf layout, sorted[150,165,180,260] · [275,285,310] · [330,340,355] · [410,420] · [440,520,545,560] — five pages, still fully sorted left to right.
total page writes16 inserts + 4 splits = 20 page writes to disk over the whole rush.
Range query, amount>400: 6 matches (410,420,440,520,545,560), and because the leaves are sorted, they sit in exactly the last 2 of 5 pages. The query walks straight there and stops.
LSM-Tree: same 16 inserts, memtable capacity 4
flush on full, no reordering
Same amounts, same arrival order, but no sorting on write — just append to memory and flush every 4th insert.
flushesExactly 16÷4 = 4 flushes, each writing one immutable SSTable in a single sequential burst.
what's inside eachSSTable 0: {340,180,260,560} · SSTable 1: {310,420,150,275} · SSTable 2: {355,440,520,165} · SSTable 3: {330,410,545,285} — each one spans a wide, overlapping amount range, because arrival order has nothing to do with amount.
total write operations4 sequential flush writes for all 16 inserts — a fifth of the B-tree's page-write count, because writes are batched instead of touched one at a time.
Range query, amount>400: every single SSTable's range overlaps 400 — even keeping per-file min/max statistics doesn't rule any of them out. The query must open all 4 of 4 SSTables to be sure it has every match.
See Both Structures, Run the Same Query
Where does "amount > 400" have to look?
same 16 inserts, two structures
Five B-tree leaf pages, four LSM SSTables — both hold the same 16 keys.
why the LSM side isn't just "worse"

The 20-vs-4 write comparison looks damning for the B-tree, and at genuinely large write volumes it is: CityCourier at citywide scale (46,080 orders/day, Section 2's hero panel) is a write-dominated workload, exactly what LSM-trees are built for — Cassandra, RocksDB and HBase all use one underneath. The read cost above is real, but it's mitigated in practice with Bloom filters (a compact structure that can quickly say "this key is definitely NOT in this SSTable," letting most files be skipped without a full read) and periodic compaction that merges old SSTables back into fewer, larger, re-sorted ones.

pitfalls
  • "LSM-trees are slower, full stop." They're slower for this specific unindexed-range-query shape. Point lookups by exact key, with a Bloom filter per SSTable, can be nearly as fast as a B-tree's, and writes are unambiguously faster.
  • "Compaction is optional cleanup." Without it, the number of SSTables a read must check only grows, and so does wasted space from old, overwritten versions of the same key. Compaction is load-bearing, not housekeeping.
  • "A B-tree's sorted order is free." It's paid for on every write, in the form of page splits and the random disk I/O of updating a page in place — the 20 page-writes above versus the LSM's 4. Sorted-on-disk and cheap-to-write are close to opposite goals.
Practice
  1. If the query were "find the order with amount exactly 355" (a point lookup, not a range) instead, how many B-tree pages need checking, and does the LSM-tree's disadvantage shrink or vanish?
    Solution

    The B-tree still needs just 1 page — 355 sits alone in the [330,340,355] leaf. The LSM-tree's situation improves sharply with Bloom filters: a filter per SSTable can rule out the 3 SSTables that don't contain 355 in roughly constant time each, leaving only 1 real read (SSTable 2, which holds it) — the same 1-page answer as the B-tree, for point lookups specifically.

  2. Head office's query changes to "amount between 150 and 200" instead of "over 400." Recompute which B-tree pages match.
    Solution

    Values in [150,200]: 180 and 165 and 150 — all three sit in the very first leaf page, [150,165,180,260]. Just 1 of 5 pages, even better than the >400 example, because this range happens to fall entirely inside one leaf's boundaries.

  3. After the rush, a compaction merges all 4 SSTables into 1 large sorted file. Redo the "amount>400" query. What's the new cost, and what did compaction actually buy?
    Solution

    One SSTable now holds all 16 keys, sorted. The query touches that single file — but unlike the B-tree's targeted 2-of-5 pages, a naive scan of one large sorted SSTable for a range still means finding the right starting offset within it (which a well-built SSTable supports via an internal index, closing most of the remaining gap with the B-tree). Compaction buys back most of the read advantage the B-tree had, at the cost of the compaction work itself happening in the background.

  4. Section 2.1 measured 20 B-tree page writes against 4 LSM flush writes for the same 16 inserts. If CityCourier scaled to 4,608,000 orders in a day (100× the citywide figure), which structure's write cost grows faster, and why does that matter for choosing a storage engine at scale?
    Solution

    The B-tree's page-write count doesn't scale linearly and predictably — every insert can trigger a split, and splits compound as the tree deepens, meaning write cost per insert can actually increase as the dataset grows. The LSM-tree's flush count scales exactly linearly (inserts ÷ memtable capacity, always), and every flush is a cheap sequential write regardless of how large the dataset has grown. At the volumes distributed storage exists for, that predictable, flat-scaling write cost is precisely why write-heavy big-data systems default to LSM-based engines.

2.6 · QUERY EXECUTION

The Same Question, Three Storage Layouts

"What did each hub earn?" is one query. How expensive it is to answer depends entirely on which of this chapter's partitioning schemes CityCourierDB actually used — and the answer isn't the same for every question.

A query engine that can't touch data directly has one real move: scatter-gather — send a piece of the question to every node that might hold relevant rows, then combine the partial answers centrally. How much network traffic that costs depends on whether the partitioning scheme happens to line up with what the query is asking for.

Worked Example · Full Aggregate vs Selective Filter

Revenue by hub, recomputed directly from all 16 rows: Hub N ₹2,085, Hub S ₹1,695, Hub E ₹1,765 — total ₹5,545, exactly matching Section 1.1.

Scatter-gather under three partitioning choices
same data, same query, different cost
Query:
Every hub's revenue needs every row, so all three nodes are contacted no matter how the data is split.

The full aggregate touches all three nodes under any partitioning scheme, because it needs every row regardless. The selective query is where the partitioning choice actually shows up in the bill: hub-based partitioning answers "Hub E only" with 1 network round trip; Section 2.3's hash-of-order-id partitioning needs 3 — the coordinator has no way to know in advance that Hub S's server holds none of Hub E's rows, so it has to ask anyway and get an empty answer back. Same final number, ₹1,765, three times the network cost to get it.

partitioning key is a query-pattern decision, not a physical fact

Nothing about the data requires partitioning by hash of order id. That choice was made in Section 2.3 because key-based point lookups ("give me order #4") were the goal. If CityCourierDB's dominant query pattern were instead "give me one hub's numbers," partitioning by hub from the start would have been the better call — at the cost of making point lookups by order id need a scatter-gather instead. There is no partitioning scheme that's cheapest for every query; there's only the one that matches what you actually ask most often.

Secondary Indexes: Finding the Needle Without Scanning the Haystack

One more query, one more callback: "find any order with a negative delivery time" — Section 1.1's veracity bug, order #10 at −60 seconds.

without an index: up to 16 row-checks, worst case — the engine has no way to know where the bad row is until it looks with a secondary index on delivery_time (Section 2.5's B-tree, keyed on a different column this time): the index is sorted ascending; −60 is the SMALLEST value in the entire dataset → it is the very first entry — found in 1 index read, no row-by-row scanning at all

A secondary index is simply an index built on a column that isn't the primary key or partition key — here, delivery_time rather than order_id. It costs extra storage and extra write work (every insert now updates two structures, not one) to buy exactly this: a query on a non-key column stops being a full scan.

pitfalls
  • "Scatter-gather is inherently slow." It's exactly as slow as the slowest node it waits on, and exactly as expensive as the number of nodes actually contacted — both of which the partitioning scheme controls. The pattern itself isn't the cost; the mismatch between partitioning and query shape is.
  • "An index makes every query faster." It makes queries on the indexed column faster, at the cost of slower writes (every index is one more structure to update) and extra storage. Indexing a column nobody queries by is pure cost with no matching benefit.
  • "More partitioning schemes is strictly more flexible." Every additional scheme (a block layout, a hash ring, a secondary index) is a structure that has to be kept correct on every write. CityCourierDB running all of this chapter's schemes at once means every order write touches blocks, a ring position, quorum replicas, a B-tree or LSM index, and a secondary index — real systems pick the subset that matches their actual query load, not all of it by default.
Practice
  1. Under hash-of-order-id partitioning, how many network round trips does "Hub N's revenue only" need, and what's the answer?
    Solution

    3 round trips — same reasoning as Hub E: the coordinator can't know in advance which nodes hold Hub N's rows under hash partitioning, so it asks all three. The answer is ₹2,085, gathered from whichever nodes' partial sums are non-zero.

  2. CityCourier's most common query becomes "give me this one order by id" (a point lookup), run far more often than any hub-level aggregate. Which partitioning scheme from this chapter should it prioritise, and why?
    Solution

    Section 2.3's hash-of-order-id ring — it turns "find order #4" into a single computed lookup (hash it, go straight to the owning node) with zero scatter-gather at all, which is precisely the access pattern consistent hashing was built for. Hub-based partitioning would force even a point lookup by id to check every node, since id has no relationship to hub.

  3. Order #10's veracity bug was found via a secondary index on delivery_time in 1 read. Would that same index have helped find order #10 by searching on amount (₹440) instead?
    Solution

    No — an index is built on one specific column's values. A delivery_time index is sorted by delivery_time and gives no shortcut for an amount-based search; that would need its own separate secondary index on amount (exactly Section 2.5's B-tree, which was keyed on amount specifically). Every column you want fast lookups on needs its own index.

  4. Combine Sections 2.4 and 2.6: if CityCourierDB uses N=3, W=2, R=2 replication (Section 2.4) and hash-of-order-id partitioning (Section 2.3), how many total network requests does a fully consistent read of a single order by id require in the worst case, and why?
    Solution

    Up to 3: the coordinator computes the order's ring position (no network cost, just arithmetic) to find which node is the primary owner — that's the 1 request a plain point lookup needs. But satisfying R=2 means checking a second replica of that same key too, since quorum reads compare multiple copies rather than trusting one. Point lookups get the ring's O(1) routing benefit; the replication factor still costs its own round trips on top, because routing and consistency are two separate concerns this chapter kept deliberately distinct.

CLOSING · CHEAT SHEET

Chapter 2, One Page

2.1 · Blocks & Replication

NameNode metadata only · DataNode holds actual blocks

Real default: 128MB blocks, replication factor 3

Heartbeat + block report → NameNode re-replicates on failure

2.2 · NoSQL Models

Key-Value opaque, exact-key only · Document nested, self-describing

Column-family sparse at scale · Graph cheap one-hop traversal

Pick the model the dominant query needs, not the trendiest one

2.3 · Partitioning & Hashing

Naive hash mod N rehashes almost everything when N changes

Consistent hashing moves only ~k/n keys on add/remove

Storage location ≠ business meaning of the key

2.4 · Replication & Quorums

W+R > N → every read quorum guaranteed to overlap every write quorum

N=3, W=2, R=2 is the common balanced default

Quorum overlap ≠ conflict resolution — that's a separate mechanism

2.5 · B-Tree vs LSM-Tree

B-Tree in-place, sorted, cheap targeted reads, costly random writes

LSM-Tree append-only, cheap sequential writes, costly unindexed reads

Bloom filters + compaction close most of the LSM read gap

2.6 · Query Execution

Full aggregates touch every node under any partitioning scheme

Selective queries only skip nodes if the partition key matches the filter

Secondary index turns "scan everything" into "read one entry"

CLOSING · SELF-TEST

Mixed Review

Eight questions, deliberately out of section order.

  1. A DataNode stops sending heartbeats. What does the NameNode need to do, and what information does it use to decide?
    Solution

    It checks its own metadata for which blocks that DataNode was holding, finds any now under-replicated, and issues copy commands to other live DataNodes to restore the replication factor — exactly Section 2.1's Hub S failure. It never touches file content directly; it only ever reasons about the block-to-node map it already keeps.

  2. A dataset has N=4 replicas. Give one (W,R) pair that guarantees quorum overlap and one that doesn't.
    Solution

    Guarantees overlap: W=3,R=2 (3+2=5>4), or W=R=3. Doesn't guarantee it: W=2,R=2 (2+2=4, not >4) — a real, common trap, since W=R=2 looks balanced but fails the strict inequality at even N.

  3. A store keeps a customer's order history as a nested JSON object with an optional "gift_note" field present on only a few orders. Which NoSQL model is this, and what would the same sparse field cost in a fixed-schema table?
    Solution

    Document model. In a fixed relational table, every row would need a "gift_note" column whether or not it's used — typically stored as NULL for the rows without one, which at scale is real wasted space for a field almost nobody fills in.

  4. Five servers sit on a consistent-hash ring holding 500 keys total, evenly spread. Roughly how many keys move when a sixth server joins?
    Solution

    Roughly k/n_new = 500/6 ≈ 83 keys — only the keys in the range between the new server and its immediate clockwise predecessor. The other ~417 keys keep their existing owner untouched.

  5. Why does a range query on a non-key column tend to be more expensive on an LSM-tree than a B-tree, even though both can answer it eventually?
    Solution

    A B-tree keeps everything sorted by that key at all times, so a range sits in a small contiguous run of pages. An LSM-tree's SSTables are ordered by flush time, not by the queried column's value, so a range's matches are smeared across every SSTable — each one has to be checked (or ruled out via stored min/max bounds) rather than walked to directly.

  6. A query asks for the total row count across an entire dataset. Does the choice of partitioning scheme change how many nodes must be contacted?
    Solution

    No — a full-dataset aggregate needs every row regardless of how it's split, so it touches every node under any partitioning scheme. Partitioning choice only changes cost for selective queries that filter to a subset, exactly Section 2.6's Hub N-vs-Hub E-only comparison.

  7. A graph database is asked to sum a numeric property across a million nodes. Is this the kind of query graph databases are built to make cheap?
    Solution

    No. Graph databases are optimised for traversal — following relationships a hop or several hops at a time — not bulk aggregation across every node, which is closer to a column-family or document store's strength. Asking a graph store to do this means paying its per-node relationship overhead for a query that never uses a single edge.

  8. Order #10's veracity bug (Section 1.1) was found via a secondary index in this chapter. Which earlier chapter first flagged that this same row was a problem, and what kind of problem was it there?
    Solution

    Section 1.1, Chapter 1 — where it was introduced as a Veracity violation (a wrong-but-present value, not a missing one). This chapter reuses the identical row as a storage-and-retrieval example: the same bad data point, viewed first as a data-quality problem and later as an indexing problem.

FURTHER READING

If You Want the Long Version

  • Hadoop: The Definitive Guide — Tom White. The full HDFS architecture behind Section 2.1's NameNode/DataNode sketch, including federation, high availability, and the actual write pipeline this chapter simplified away.
  • Designing Data-Intensive Applications — Martin Kleppmann. Covers Sections 2.2 through 2.4 in real depth — data model trade-offs, partitioning strategies beyond consistent hashing, and replication consistency models with more nuance than N/W/R alone.
  • Mining of Massive Datasets — Leskovec, Rajaraman & Ullman. Its index-structures chapter goes further into the B-Tree/LSM-Tree family than Section 2.5 had room for, including the hashing schemes distributed databases build on.

Chapter 2 of 7 · CSUE301 Big Data Analytics · builds on Chapter 1's CAP theorem (Section 1.6) and data lifecycle (Section 1.2).
Next → Chapter 3, Parallel & Distributed Data Processing.