Sources and References

Primary textbooks — Jimmy Lin and Chris Dyer, Data-Intensive Text Processing with MapReduce, Morgan & Claypool, 2010 (Chapters 1–6, freely available); Tom White, Hadoop: The Definitive Guide, 4th ed., O’Reilly, 2015. Supplementary texts — Holden Karau and Rachel Warren, Learning Spark, 2nd ed., O’Reilly, 2020; Marc Peter Deisenroth, A. Aldo Faisal, and Cheng Soon Ong, Mathematics for Machine Learning, Cambridge University Press, 2020. Online resources — Apache Spark documentation (spark.apache.org); Apache Kafka documentation (kafka.apache.org); Google Bigtable paper (Chang et al., OSDI 2006); Google MapReduce paper (Dean & Ghemawat, OSDI 2004); Apache HBase documentation; Flink documentation (flink.apache.org).

Chapter 1: The Big Data Problem

The Data Deluge

The premise of this course is simple: the amount of data produced by the modern world has vastly outpaced the ability of a single machine to process it. Social media platforms generate terabytes of logs per hour. Genomic sequencing produces petabytes of raw reads per year. IoT sensors in industrial plants emit continuous streams from millions of devices. The Large Hadron Collider generates roughly 15 petabytes of data annually. Scientific simulations in climate modeling and astrophysics routinely produce outputs too large for any one disk.

The fundamental problem is not merely storage — hard drives have grown cheaper — but processing. A single machine with 32 cores and 256 GB of RAM can scan perhaps 10 GB/s from local NVMe storage. At that rate, scanning 1 petabyte takes over 27 hours. And that assumes the data fits on local disks, the algorithm is purely sequential, and there is no memory pressure.

Scaling Up vs. Scaling Out

Two architectural responses exist to this problem. Scaling up (vertical scaling) means purchasing a larger machine: more cores, more RAM, faster storage. This works until it doesn’t — the largest commodity servers top out around 6 TB of RAM and 128 cores, and specialized hardware (SGI UV, for instance) becomes astronomically expensive. More importantly, a single machine remains a single point of failure.

Scaling out (horizontal scaling) means distributing the work across many commodity machines connected by a network. A cluster of 1,000 nodes, each with 10 cores and 64 GB RAM, provides 10,000 cores and 64 TB of memory at a fraction of the cost of an equivalently powerful single machine. The catch is programming complexity: the developer must now reason about coordination, communication, and failure modes that simply do not exist on a single machine.

The Distributed Systems Challenge

Distributed systems introduce a set of failure modes that fundamentally change how software must be written. The “fallacies of distributed computing” (attributed to Peter Deutsch and colleagues at Sun Microsystems) enumerate common false assumptions:

Beyond the network, distributed systems must cope with partial failures — a scenario with no analogue on a single machine. When you call a remote procedure and receive no response, you cannot tell whether the message was lost, the remote node crashed, the response was lost, or the remote node is running slowly. This ambiguity forces developers to design for idempotency and re-execution.

Fault tolerance is achieved through several complementary mechanisms: redundancy (store multiple copies of data so no single disk failure causes data loss); replication (replicate computation state across nodes so any node can take over); heartbeats (nodes send periodic liveness signals, and the absence of a heartbeat within a timeout triggers a failure declaration); and re-execution (when a task fails, simply re-run it on another node — this is only possible if tasks are deterministic and side-effect-free, which MapReduce enforces by design).

The Google Stack

The modern era of large-scale data processing was defined by three papers from Google, published between 2003 and 2006:

1. GFS (Ghemawat, Gobioff, Leung — SOSP 2003): The Google File System. A distributed file system optimised for large sequential reads and append-only writes, running on commodity hardware with automatic replication and failure recovery.
2. MapReduce (Dean and Ghemawat — OSDI 2004): A programming model and execution engine for parallel data processing, hiding all distributed systems complexity behind a two-function interface.
3. Bigtable (Chang et al. — OSDI 2006): A distributed storage system for structured data, providing random reads and writes at scale on top of GFS.

These three papers inspired the open-source Hadoop ecosystem: HDFS (GFS clone), Hadoop MapReduce (MapReduce clone), and HBase (Bigtable clone).

HDFS Architecture

HDFS (Hadoop Distributed File System) is the storage layer that underpins the Hadoop ecosystem. Its design is deliberately simple and optimised for the MapReduce access pattern: write once, read many, in large sequential scans.

Files are divided into blocks of 128 MB by default. Each block is replicated to three DataNodes using a rack-aware placement strategy: one replica on the local rack, two on a separate rack. This means the cluster can tolerate the failure of any single node or any single rack without data loss.

The NameNode is the metadata master. It holds the entire file system namespace — the mapping from file paths to block IDs, and from block IDs to the DataNodes that hold each replica — entirely in memory for fast access. This is both a strength (O(1) metadata lookups) and a weakness (the NameNode is a single point of failure and a scalability bottleneck; HDFS Federation addresses this by sharding the namespace across multiple NameNodes).

Reads proceed as follows: the client contacts the NameNode to learn which DataNodes hold the needed blocks, then reads directly from the closest DataNode (locality is inferred from network topology). Writes use a pipeline: the client sends data to DataNode 1, which forwards to DataNode 2, which forwards to DataNode 3; acknowledgements flow back along the pipeline. HDFS does not support random updates — files are write-once (or append-only). This constraint, which would be unacceptable in a general-purpose file system, is exactly what allows HDFS to achieve high throughput: there are no update locks, no cache invalidation, no complex consistency protocols.

Chapter 2: MapReduce Algorithm Design

The Programming Model

MapReduce reduces the problem of parallel data processing to two user-supplied functions:

The framework handles everything in between: reading data from HDFS, launching map tasks across the cluster, partitioning and sorting intermediate key-value pairs by key (the shuffle), grouping all values for a given key, launching reduce tasks, and writing output back to HDFS. The developer only writes the map and reduce functions.

The canonical example is word count. The mapper tokenises each line and emits (word, 1) for every token. The framework groups all (word, 1) pairs by word. The reducer sums the list of ones to produce the total count. Despite its simplicity, word count captures the essential structure of a vast class of distributed aggregation problems.

The execution timeline is: (1) split input into chunks and assign one map task per chunk; (2) mappers run in parallel, each writing intermediate output to local disk; (3) the shuffle phase sorts and partitions intermediate records by key, then transfers them across the network to reducers; (4) reducers run in parallel, each processing all values for a contiguous range of keys; (5) output is written to HDFS.

Combiners

The shuffle is the most expensive phase of MapReduce because it involves network I/O, which is orders of magnitude slower than local disk I/O. The combiner is a mini-reducer that runs on each mapper’s local output before the shuffle, reducing the number of key-value pairs that must traverse the network.

For word count, the combiner is identical to the reducer: it sums the partial counts from one mapper before they leave the node. If a mapper processes a 128 MB chunk containing the word “the” 50,000 times, without a combiner it emits 50,000 ("the", 1) pairs. With a combiner, it emits one ("the", 50000) pair — a 50,000× reduction in network traffic for that key.

The critical constraint: the combiner must be commutative and associative, because it may be applied zero, one, or multiple times — the framework makes no guarantee. This rules out operations like average (you cannot average partial averages without knowing the counts), but the workaround is to emit (sum, count) pairs and compute the final average in the reducer.

Partitioners

The partitioner determines which reducer receives each intermediate key. The default is a hash partitioner: hash(key) mod numReducers. This distributes keys uniformly in expectation but can produce data skew if the key distribution is non-uniform (e.g., if 90% of your data has the same key, one reducer gets 90% of the work).

Custom partitioners solve two classes of problems. First, load balancing: if you know the key distribution in advance (or sample it), you can partition to equalise reducer work. Second, total order: HDFS output files are sorted within each reducer’s partition, but not across partitions. To produce a globally sorted output, use TotalOrderPartitioner, which samples the key space and assigns contiguous key ranges to reducers.

Design Pattern: Local Aggregation (In-Mapper Combining)

The combiner helps, but it operates on the serialised output of the mapper and requires a separate pass over that output. In-mapper combining goes further: the mapper maintains an in-memory associative array, accumulating partial aggregates across all key-value pairs it processes, and only emits at the end of the map task (or periodically to bound memory usage).

For word count, the mapper holds a HashMap<String, Integer> and increments counts in memory. At the end of the task, it emits one pair per unique word seen. This is faster than the combiner because it avoids serialisation of intermediate output entirely.

The trade-off is memory: the in-memory array can grow without bound if the key space is large. For word count over natural language text, the vocabulary is bounded (perhaps 1 million unique words), so the array stays manageable. For problems with unbounded key spaces, in-mapper combining must emit periodically.

Design Pattern: Pairs and Stripes for Co-occurrence

Computing word co-occurrence matrices (how often do words appear together in a window?) illustrates a fundamental design choice between two MapReduce patterns.

Pairs approach: for each word \(w_i\) in a window, emit ((w_i, w_j), 1) for every co-occurring word \(w_j`. The reducer counts how many times each pair occurs. To compute the marginal \(P(w_i)\), emit a special marker ((w_i, *), 1) for each occurrence of \(w_i`. Problem: the number of intermediate pairs is \(O(n \cdot k)\) where \(k\) is window size, and each pair traverses the network separately.

Stripes approach: for each word \(w_i\), emit (w_i, {w_j: 1, w_k: 1, ...}) — a single key with a hash map as value. The reducer merges hash maps by summing counts. Much less network traffic (one record per anchor word per mapper instead of one per co-occurring pair), but each value is a large object that may stress the serialisation layer.

The choice depends on the vocabulary size and cluster characteristics. For small vocabularies, stripes dominates. For large vocabularies where the per-word hash map would exceed memory, pairs becomes more practical.

Design Pattern: Order Inversion

Computing relative frequencies \(P(w_j | w_i) = \text{count}(w_i, w_j) / \text{count}(w_i)\) requires knowing the marginal \(\text{count}(w_i)\) before processing individual pair counts. Order inversion exploits MapReduce’s sorting to ensure the marginal arrives at the reducer before the joint counts.

The trick: emit ((w_i, *), 1) for the marginal with a special sentinel key * that sorts lexicographically before all other values of \(w_j\). The reducer sees the marginal first, stores it, then processes subsequent pair counts to compute relative frequencies. This converts what would require two MapReduce passes into one.

Design Pattern: Value-to-Key Conversion (Secondary Sort)

MapReduce sorts records by key before delivering them to reducers. By pushing part of the value into the key, you can impose a sort order on the values within a group.

For example, to find the maximum temperature per weather station, the natural approach emits (station_id, temperature) and the reducer computes the max. But if you need to emit records in temperature-sorted order within each station, encode (station_id, temperature) as the composite key and define a custom partitioner that partitions only on station_id (so all records for a station go to the same reducer) and a custom comparator that sorts by the full composite key. The reducer then receives records already sorted by temperature.

Limitations of MapReduce

MapReduce’s rigid two-stage structure is both its strength and its weakness. It is easy to reason about, easy to make fault-tolerant (re-run any failed task), and easy to schedule. But many algorithms are inherently iterative: PageRank, gradient descent, BFS, k-means clustering. Each MapReduce iteration reads from HDFS and writes back to HDFS — a full round-trip through slow distributed storage. An algorithm requiring 50 iterations incurs 50 HDFS reads and 50 HDFS writes. Spark was built to fix this.

A second limitation is the rigidity of the two-stage model for expressing multi-step pipelines. A SQL-style query joining three tables, filtering, and aggregating naturally maps to a pipeline of five or six operators; in MapReduce, each join requires a separate MapReduce job, so a three-table join requires at least two jobs with intermediate HDFS materialisation. Hive and Pig addressed this by compiling high-level query languages to chains of MapReduce jobs, but the materialisation overhead remained. Spark’s DAG execution model eliminates this by allowing arbitrary chains of transformations to execute in a single pass without materialising intermediate results to distributed storage.

A third limitation is the absence of a query optimizer. MapReduce provides no cost model, no statistics, no plan enumeration. The developer must manually choose partition counts, join strategies, and combiner usage — decisions that a relational optimizer makes automatically. This makes MapReduce programs brittle: a program tuned for 100 GB of input may perform poorly on 10 TB of input because the manually chosen partition count is no longer appropriate. SparkSQL’s Catalyst optimizer and AQE (Adaptive Query Execution) represent the convergence point: a system with the flexibility of MapReduce’s arbitrary code execution and the cost-awareness of a relational query optimizer.

Chapter 3: From MapReduce to Spark

Resilient Distributed Datasets

The core abstraction in Spark is the Resilient Distributed Dataset (RDD). An RDD is an immutable, partitioned collection of records distributed across a cluster. Unlike MapReduce’s fixed two-stage model, RDDs support a rich set of transformations that compose into arbitrary DAGs (directed acyclic graphs) of computation.

The “resilient” in RDD refers to its fault tolerance mechanism. Instead of replicating data (expensive), Spark records the lineage of each RDD — the sequence of transformations that produced it from the original data source. If a partition is lost due to node failure, Spark can recompute it by replaying the lineage from the last checkpoint or the original data. This is efficient because lineage graphs are typically shallow (a few transformations deep), and recomputation is embarrassingly parallel.

RDDs are lazily evaluated: calling a transformation (e.g., map, filter) does not trigger computation. It merely adds a node to the lineage DAG. Computation only begins when an action is called (e.g., count, collect, saveAsTextFile). This allows Spark’s optimizer to fuse transformations, eliminating intermediate materialisation.

Transformations and Actions

Transformations build the DAG; actions execute it.

Spark Execution Model

When an action is called, Spark submits the lineage DAG to the DAG scheduler. The DAG scheduler analyses the graph and partitions it into stages at shuffle boundaries. Within a stage, all transformations can be pipelined — a record flows from one transformation to the next without materialisation. Across stage boundaries, a full shuffle (analogous to MapReduce’s shuffle/sort) must occur.

Narrow dependencies occur when each output partition depends on at most one input partition (e.g., map, filter). These can be pipelined within a stage. Wide dependencies occur when an output partition depends on multiple input partitions (e.g., groupByKey, join). These require a shuffle and create stage boundaries.

The task scheduler takes each stage and launches tasks on executors. An executor is a JVM process running on a worker node, holding a fixed number of CPU cores and a fixed memory budget. Tasks within a stage run in parallel across partitions.

text = sc.textFile("hdfs://data/")           # RDD 1: raw lines
words = text.flatMap(lambda line: line.split()) # RDD 2: words (narrow)
pairs = words.map(lambda w: (w, 1))           # RDD 3: (word, 1) pairs (narrow)
counts = pairs.reduceByKey(lambda a, b: a+b)  # RDD 4: (word, count) — SHUFFLE
sorted_counts = counts.sortByKey()            # RDD 5: sorted — SHUFFLE
sorted_counts.saveAsTextFile("hdfs://output/")

Persistence and Caching

The killer feature of Spark for iterative algorithms is caching. Calling rdd.cache() (equivalent to rdd.persist(StorageLevel.MEMORY_ONLY)) instructs Spark to materialise the RDD in executor memory after it is first computed, rather than recomputing it from lineage on each subsequent use.

For PageRank running 30 iterations, the adjacency list is read once from HDFS and cached. Each iteration reuses the cached RDD rather than re-reading from disk. This alone can yield 10–100× speedups over equivalent MapReduce implementations.

Storage levels offer trade-offs:

Pair RDDs and Key-Value Operations

reduceByKey is almost always preferable to groupByKey for aggregation tasks. groupByKey collects all values for a key into a single in-memory list before passing it to the user function — if a key has millions of values, this can cause OOM errors. reduceByKey applies a binary associative function incrementally and can perform local aggregation on each partition before the shuffle, exactly like a combiner.

aggregateByKey(zeroValue)(seqOp, combOp) generalises reduceByKey: seqOp merges a new value into an accumulator, combOp merges two accumulators. This supports cases where the accumulator type differs from the value type (e.g., computing a per-key mean requires accumulating a (sum, count) pair).

Accumulators and Broadcast Variables

Accumulators are distributed counters or sums. Workers can only add to an accumulator; only the driver can read the total. They are useful for counting events during a transformation (e.g., counting malformed records) without returning data to the driver via a full reduce action.

Broadcast variables distribute a read-only value to all executors exactly once, rather than serialising it into each task closure. This is critical for map-side joins: if a lookup table is small enough to fit in memory (say, 100 MB), broadcast it to all workers, then each mapper does a local hash-table lookup instead of a shuffle.

Spark vs. MapReduce

Chapter 4: Analyzing Text

Inverted Indexes

An inverted index maps each term to the list of documents (and positions) in which it appears — the core data structure of every search engine. Building one at scale is a canonical MapReduce application.

The mapper processes each document, tokenises the text, and emits (term, docId) for every token. The reducer receives all (term, docId) pairs grouped by term and assembles the posting list: a sorted list of document IDs, typically augmented with term frequency \(\text{tf}(t, d)\) (how many times term \(t\) appears in document \(d\)).

For web-scale corpora, posting lists can contain billions of entries and must be compressed. Delta encoding stores gaps between successive document IDs rather than absolute IDs (since IDs are sorted, gaps are small positive integers). Variable-byte encoding encodes small integers in fewer bytes than a fixed 4-byte int. Together, these techniques can reduce posting list size by 10–20×.

TF-IDF Weighting

Raw term frequency is a poor measure of relevance — words like “the” and “is” appear frequently in all documents and carry no discriminative information. TF-IDF (term frequency–inverse document frequency) discounts ubiquitous terms:

where \(N\) is the total number of documents and \(\text{df}(t)\) is the number of documents containing term \(t\). Words that appear in many documents get a low IDF weight; rare, specific terms get high weight. Computing IDF requires knowing \(\text{df}(t)\), which is the output of one MapReduce pass (count documents per term). A second pass then computes TF-IDF scores.

Documents are represented as sparse vectors in the term space: most entries are zero (a document uses only a tiny fraction of the vocabulary). Cosine similarity between document vectors is the standard relevance metric.

Language Models

A language model assigns a probability to a sequence of words. The unigram model estimates:

This is simply the relative frequency of word \(w\) in document \(d\). For information retrieval, a document is ranked by the probability it would generate the query words under this model.

The problem is zero probability: if a query word does not appear in a document, the model assigns probability zero, causing the document to be ranked last even if it is highly relevant. Smoothing mixes the document model with a background (collection) model:

Jelinek-Mercer smoothing:

Dirichlet smoothing:

Documents are ranked by KL-divergence between the query model and the document model, or equivalently by log-likelihood of the query under the smoothed document model.

Naive Bayes Text Classification

Naive Bayes classifies documents by applying Bayes’ theorem with the strong independence assumption that words occur independently given the class:

Training consists of estimating \(P(c)\) (class prior) and \(P(w \mid c)\) (per-class word distribution) from labelled data — pure counting, trivially parallelisable in MapReduce. Prediction computes the log of the above expression to avoid floating-point underflow:

Laplace smoothing (add-one smoothing) prevents zero probability for unseen words: \(P(w \mid c) = (\text{count}(w, c) + 1) / (|V| + \sum_{w'} \text{count}(w', c))\).

Latent Dirichlet Allocation

Topic models treat documents as mixtures of latent topics and topics as distributions over words. LDA (Blei et al., 2003) is the standard topic model. Training via collapsed Gibbs sampling iterates: for each word token in the corpus, remove its current topic assignment, then resample a new topic based on how frequently the word appears in each topic and how frequently the document uses each topic.

At scale, this sampling loop over billions of word tokens is parallelised in Spark by partitioning the corpus across workers, each maintaining local topic-word counts that are periodically synchronised. The SparkML LDA estimator implements both online variational Bayes (faster) and Expectation-Maximisation variants.

Chapter 5: Analyzing Graphs

Graph Representations and Challenges

Graphs are pervasive in data-intensive applications: the web graph (pages and hyperlinks), social graphs (users and friendships), knowledge graphs (entities and relations), and citation networks. A graph with \(n\) nodes and \(m\) edges can be stored in HDFS as an adjacency list: each line contains a node ID followed by the IDs of its neighbours. For sparse graphs (most real-world graphs), this is far more compact than an adjacency matrix.

Processing graphs in MapReduce is fundamentally awkward because graph algorithms are typically iterative (propagate information along edges until convergence) and inherently non-local (a single step may require accessing the neighbours’ neighbours’ neighbours). Each MapReduce iteration must load the entire graph from HDFS, limiting iteration throughput.

Parallel BFS in MapReduce

Breadth-first search computes the shortest-hop distance from a source node \(s\) to all other nodes. In MapReduce, the standard approach maintains a distance array and iterates:

1. Represent each node as a record (node_id, current_distance, adjacency_list).
2. Mapper: if current_distance < ∞, emit (neighbour, current_distance + 1) for each neighbour; also emit the node itself to preserve the adjacency list.
3. Reducer: for each node, take the minimum distance seen across all emitted records.
4. Repeat until no distances change (detected via a counter accumulator).

The inefficiency is that every node is processed every iteration, even nodes far from the current frontier. Spark’s caching helps (the adjacency list is cached), but the fundamental issue of scanning the full graph per iteration remains. Pregel (Chapter 10) addresses this with message-passing that only activates nodes on the frontier.

PageRank

PageRank models a random surfer who follows hyperlinks at random with probability \(d\) and teleports to a random page with probability \(1 - d\):

This is solved by power iteration: initialise \(P(v) = 1/N\) for all \(v\), then repeatedly apply the above formula until the \(L_1\) norm of the change vector falls below a threshold (typically 10–30 iterations suffice).

The MapReduce implementation: the mapper emits (neighbour, P(u) / out(u)) for each outgoing edge, distributing the source’s rank to all its neighbours. It also emits the adjacency list itself to reconstruct the graph. The reducer sums incoming contributions and applies the damping factor.

Dangling nodes (pages with no outgoing links) accumulate rank and never distribute it, causing rank to “leak” from the system. The fix is to add a teleportation term that distributes dangling node mass uniformly across all pages.

In Spark, PageRank benefits enormously from caching the adjacency list RDD. The RDD containing links is read from HDFS once and cached, while the ranks RDD is updated each iteration:

val links = sc.textFile(…).map(…).cache()
var ranks = links.mapValues(_ => 1.0)
for (_ <- 1 to numIter) {
  val contribs = links.join(ranks).flatMap { case (_, (urls, rank)) =>
    urls.map(url => (url, rank / urls.size))
  }
  ranks = contribs.reduceByKey(_ + _).mapValues(0.15 + 0.85 * _)
}

Graph Partitioning

For distributed graph processing, how nodes and edges are partitioned across machines fundamentally affects communication cost. Edge-cut partitioning places vertices on partitions and cuts edges that cross partitions; vertex-cut partitioning places edges on partitions and replicates vertices. For power-law graphs (most real social and web graphs), vertex-cut partitioning typically yields fewer replicas because high-degree vertices would otherwise cause massive edge cuts.

Chapter 6: Data Mining and Machine Learning at Scale

Gradient Descent

Most supervised machine learning reduces to minimising a loss function \(\mathcal{L}(\theta)\) over model parameters \(\theta\). Gradient descent updates:

where \(\eta\) is the learning rate. The loss decomposes across training examples:

Batch gradient descent sums gradients over the entire dataset before updating — parallelisable in MapReduce (each mapper computes gradients on its shard; the reducer sums them) but slow to converge because it takes one update per full pass over the data.

Stochastic gradient descent (SGD) updates after every example — fast convergence but noisy. Mini-batch GD processes small batches (32–256 examples) per update, balancing convergence speed and gradient noise.

MapReduce Gradient Computation

The MapReduce pattern for gradient computation is straightforward:

1. The driver broadcasts the current parameter vector \(\theta\) to all workers.
2. Each mapper computes the gradient contribution from its local data shard.
3. The reducer sums gradients across all shards.
4. The driver applies the update and repeats.

Communication cost per iteration is \(O(|\theta|)\) — the parameter vector must traverse the network each iteration. For deep learning models with billions of parameters, this becomes the bottleneck.

Logistic Regression

Logistic regression models binary classification with the sigmoid function \(\sigma(z) = 1 / (1 + e^{-z})\):

The cross-entropy loss is:

The gradient is:

This sum is embarrassingly parallel. In practice, distributed logistic regression uses L-BFGS (limited-memory Broyden-Fletcher-Goldfarb-Shanno), a quasi-Newton method that approximates the inverse Hessian and converges in far fewer iterations than SGD, making the communication overhead worthwhile.

Parameter Server Architecture

For very large models, the driver bottleneck is addressed by the parameter server (Li et al., 2014). The parameter space is sharded across a set of dedicated server nodes. Worker nodes pull the parameters they need, compute gradients on local data, and push gradients back. Servers apply updates asynchronously.

Asynchronous SGD introduces stale gradients: a worker may compute a gradient using parameter values that are several updates old by the time the gradient reaches the server. Surprisingly, asynchronous SGD still converges for many convex and non-convex problems — the staleness adds noise but does not prevent convergence, and the throughput gains from eliminating synchronisation barriers typically outweigh the convergence penalty.

Spark MLlib

Spark’s MLlib provides a Pipelines API modelled after scikit-learn:

• Transformer: applies a transformation to a DataFrame (e.g., Tokenizer, StandardScaler, PCA).
• Estimator: learns from data and produces a Transformer (e.g., LogisticRegression, KMeans).
• Pipeline: chains Transformers and Estimators into a single workflow; calling fit() on a Pipeline trains all Estimators in sequence.
• PipelineModel: the fitted Pipeline, ready for transform().

Key algorithms available: logistic regression, linear SVM, random forests, gradient-boosted trees, k-means, LDA, collaborative filtering (ALS), PCA, Word2Vec, TF-IDF feature extraction, and the hashing trick (feature hashing for dimensionality reduction of sparse features).

Distributed K-Means

K-means clustering iterates between assigning each point to its nearest centroid and recomputing centroids as the mean of assigned points. In MapReduce/Spark:

1. Initialisation: k-means++ selects initial centroids by sampling with probability proportional to squared distance from already-selected centroids, improving convergence.
2. Assignment step (map): each mapper computes the nearest centroid for each of its points, emitting (centroid_id, point).
3. Update step (reduce): each reducer receives all points assigned to one centroid and computes the new centroid as their mean.
4. Convergence: repeat until centroid movement falls below a threshold or a maximum iteration count is reached.

The broadcast variable for centroids (at most \(k \times d\) floats) is sent to all workers each iteration, avoiding a shuffle for the assignment step.

Chapter 7: Analyzing Relational Data

Relational Algebra in MapReduce

Relational operations map naturally to MapReduce. Selection (\(\sigma\)) is a map-side filter: each mapper discards records not matching the predicate, with no shuffle required. Projection (\(\pi\)) is a map-side field extraction. The expensive operations are joins (\(\bowtie\)) and group-by aggregations (\(\gamma\)).

Reduce-Side Join

The most general join strategy: tag each record with its source relation before the shuffle, then join in the reducer.

1. Mapper for relation \(R\): emit (join_key, ("R", record)).
2. Mapper for relation \(S\): emit (join_key, ("S", record)).
3. Reducer: receives all records for a key, partitioned by source tag. Compute the Cartesian product of records from \(R\) with records from \(S\).

This works for any join size but requires both relations to be shuffled in full, giving network cost \(O(|R| + |S|)\).

Map-Side Join (Broadcast Join)

If one relation is small enough to fit in memory (the “dimension table” in a star schema), broadcast it to all mappers as a hash table. Each mapper then performs a local hash lookup for each record in the large relation, with no shuffle at all.

This is the most efficient join strategy when applicable, reducing network cost to \(O(\text{small relation})\) for the broadcast plus zero shuffle. Spark implements this automatically as a broadcast hash join when one side is smaller than spark.sql.autoBroadcastJoinThreshold (default 10 MB).

Repartition Join

When neither relation fits in memory, a repartition join pre-partitions both relations on the join key into the same set of partitions. Co-located partition pairs are joined locally without a full shuffle across the network. This requires both relations to be sorted by join key within each partition.

Spark SQL and DataFrames

Spark SQL provides a structured API over RDDs. A DataFrame is an RDD of Row objects with a named schema, analogous to a relational table. The same physical execution engine backs both the DataFrame API and SQL queries.

df = spark.read.parquet("users.parquet")
df.filter(df.age > 25).groupBy("city").agg(count("*")).show()

The Catalyst optimizer applies a sequence of rule-based transformations (constant folding, predicate pushdown, projection pruning) and cost-based optimisations (join reordering based on table statistics) to the logical plan before generating the physical plan. The Tungsten execution engine generates JVM bytecode at runtime (whole-stage code generation), eliminates Java object overhead with unsafe memory management, and optimises for CPU cache locality.

Adaptive Query Execution

Spark 3.0 introduced Adaptive Query Execution (AQE), which re-optimises the query plan at runtime using statistics collected from completed stages. Key features: coalescing small shuffle partitions (avoids the overhead of thousands of tiny tasks), dynamically switching join strategies (if a table turns out to be smaller than expected, switch from sort-merge join to broadcast hash join), and skew join optimisation (split skewed partitions).

Delta Lake and the Lakehouse

Traditional data lakes (HDFS/S3 with Parquet files) lack ACID transactions: concurrent writes can corrupt files, schema changes are fragile, and “time travel” (querying historical snapshots) is impossible. Delta Lake (Databricks, open-sourced 2019) adds a transaction log (_delta_log/) to an object storage directory. The log records every operation as a JSON entry; Parquet files are never modified in place.

This enables: ACID transactions (optimistic concurrency control via log entry conflicts), schema enforcement and evolution, time travel (df.read.format("delta").option("versionAsOf", 5)), and Z-ordering (space-filling curve multi-column sort for data skipping). Delta Lake represents the “Lakehouse” architecture: the scalability and cost of a data lake with the reliability and governance of a data warehouse.

Chapter 8: Real-Time Analytics — Streaming Systems

Motivation

Batch processing (MapReduce, Spark) is inherently retrospective: you process data that has already accumulated. Many applications require processing data as it arrives, with latency measured in milliseconds to seconds rather than hours. Financial fraud detection must flag a suspicious transaction before it settles. Real-time bidding in digital advertising must decide within 100 ms whether to bid on an ad impression. IoT platforms must alert operators seconds after a sensor exceeds a threshold.

Apache Kafka

Kafka is a distributed log optimised for high-throughput, durable, ordered message delivery. Its core abstraction is the topic, a named, ordered sequence of records. Topics are partitioned — each partition is an ordered, immutable sequence of records stored on one or more brokers. Producers append records to partitions; consumers read records sequentially from a given offset.

The key insight of Kafka’s design is that the log is durable: records are retained for a configurable period (days to weeks), not deleted upon consumption. This allows consumers to replay history, multiple consumer groups to independently consume the same topic, and the stream to serve as a source of truth for downstream reprocessing.

Consumer groups enable parallel consumption: each partition is consumed by exactly one consumer within a group, so a group with \(n\) members can process up to \(n\) partitions in parallel. Adding consumers beyond the partition count yields no additional parallelism.

Stream Processing Fundamentals

Two time domains are critical in stream processing:

• Event time: the time at which an event actually occurred (embedded in the event’s data).
• Processing time: the time at which the event is processed by the stream processor.

These differ because of network delays, retransmissions, and out-of-order delivery. A sensor reading from 10:00:00 may arrive at the stream processor at 10:00:05. For most analytics applications (e.g., “how many clicks occurred in the 9am–10am window?”), event time is correct; processing time is an approximation that can be significantly off.

Watermarks are the mechanism for reasoning about event-time progress despite out-of-order data. A watermark at time \(t\) is a declaration that all events with event time \(\leq t\) have been observed. Any event arriving with event time \(< t\) after the watermark is a late event and is either dropped or handled via a special late-data policy.

Windowing divides the infinite stream into finite chunks for aggregation:

• Tumbling windows: fixed-size, non-overlapping (e.g., one window per hour).
• Sliding windows: fixed-size, overlapping by a hop (e.g., 1-hour windows updated every 10 minutes).
• Session windows: variable-size, defined by gaps in activity (e.g., a user session ends after 30 minutes of inactivity).

Spark Structured Streaming

Spark Structured Streaming extends the DataFrame API to streaming data. The same SQL and DataFrame operations work on both batch and streaming DataFrames. Internally, Spark treats the stream as an unbounded table to which rows are continuously appended, and executes micro-batches (by default, as fast as possible) against it.

Three output modes govern what is written to the sink per micro-batch:

• Append mode: only new rows are written (suitable for stateless queries or windows with watermarks).
• Complete mode: the entire result table is rewritten each batch (suitable for aggregations that may update historical rows).
• Update mode: only rows that changed since the last batch are written.

Stateful aggregations (e.g., per-session counts) use a state store (backed by RocksDB or in-memory hash maps) to maintain per-key state across micro-batches.

Apache Flink

Flink is a true stream processor: it processes events one at a time rather than in micro-batches. This gives Flink sub-millisecond latency (compared to Spark Streaming’s latency of 100ms–1s) at the cost of slightly higher throughput overhead per event.

Flink’s fault tolerance relies on distributed snapshots (inspired by the Chandy-Lamport algorithm). Periodically, a barrier is injected into each input stream. When an operator has received the barrier on all inputs, it snapshots its state to durable storage. Recovery replays the input from the last barrier. Combined with Kafka’s durable log, this provides exactly-once semantics end-to-end: no event is processed more than once, and none is lost, even after failures.

Flink’s RocksDB state backend stores keyed state in an embedded RocksDB instance on each task manager, enabling state larger than memory (important for long-running streaming applications with millions of keys).

Exactly-Once Semantics

The exactly-once guarantee is the gold standard for streaming correctness but the most expensive to achieve:

• At-most-once: messages may be lost; no duplicates. Lowest overhead.
• At-least-once: no messages are lost; duplicates possible. Medium overhead.
• Exactly-once: no messages lost, no duplicates. Highest overhead.

Kafka achieves exactly-once producer delivery via idempotent producers (sequence numbers prevent duplicate writes on retry) and transactions (atomic writes across partitions). Flink’s checkpointing mechanism, combined with Kafka transactions for output, provides end-to-end exactly-once.

Chapter 9: Mutable State — Key-Value Stores and HBase

Limitations of HDFS for Mutable State

HDFS is optimised for sequential, batch access to immutable data. It cannot support the access patterns of many real-world applications: random reads of individual records (e.g., fetch user profile by user ID), low-latency point queries (e.g., check if a URL has been crawled), or frequent small updates (e.g., increment a counter). These patterns require a different storage abstraction.

Google Bigtable

Bigtable (Chang et al., OSDI 2006) is described as a “sparse, distributed, persistent, multi-dimensional sorted map”:

The row key is an arbitrary byte string; rows are sorted lexicographically. All data for a row key is co-located, enabling efficient range scans. Column families group related columns for separate physical storage and compaction settings; they must be declared at schema creation time. Column qualifiers are arbitrary byte strings within a family; they need not be declared. Timestamps allow multiple versions of each cell.

Data is stored in tablets (ranges of the row key space), each served by one tablet server. Writes go to an in-memory memtable and a write-ahead log (WAL) for durability. When the memtable fills, it is flushed to an immutable SSTable on GFS. Reads merge the memtable and all SSTables. Periodic compaction merges SSTables, reclaiming space and reducing read amplification.

Apache HBase

HBase is the open-source Bigtable clone that runs on HDFS. The architecture mirrors Bigtable: Region servers serve data for a contiguous range of row keys; each region has a MemStore (in-memory write buffer), a WAL (HLog, stored on HDFS for durability), and a set of HFiles (immutable SSTables on HDFS). A central HMaster manages region assignment but is not in the data path.

LSM Trees

The data structure underlying both Bigtable and HBase is the Log-Structured Merge Tree (LSM tree). Writes are directed to an in-memory buffer (L0), which is fast. When L0 is full, it is flushed to an on-disk sorted structure (L1 SSTable). Periodically, SSTables at level \(i\) are merged and pushed to level \(i+1\) in a compaction process.

This design separates the write path (sequential, in-memory) from the read path (may require checking multiple levels). The trade-off:

RocksDB (Facebook’s LSM implementation) is the state backend for many modern systems including Flink, Kafka Streams, and TiKV.

Row Key Design

The row key is the only primary index in HBase. It determines:

• Co-location: rows with adjacent keys are co-located in the same region, enabling efficient range scans.
• Load distribution: if all writes go to the same lexicographic range (e.g., using a timestamp prefix means all recent writes go to the last region), one region server becomes a hotspot.

Good row key design avoids hotspots. For time-series data, salting prepends a random prefix to distribute writes across regions. Hash bucketing replaces the natural key with hash(key) + key, distributing rows uniformly at the cost of losing range-scan capability.

The Google Bigtable Design for Web Search

Bigtable’s data model is elegantly suited to web search infrastructure. Consider the web crawl table used by Google: the row key is the reversed hostname plus path (e.g., com.google.www/search). Reversing the hostname groups all pages from the same domain together lexicographically, enabling efficient range scans over all pages from google.com. This is a direct example of the row key design principle: choose the key to co-locate data that is frequently accessed together.

Column families separate different types of data: a contents: family stores the raw HTML of each page (one cell per crawl timestamp, allowing the history of a page’s content to be retrieved); a anchor: family stores all anchor texts pointing to this page (each column qualifier is the URL of the referring page, so anchor:cnn.com/ stores the anchor text that CNN’s homepage uses when linking to this page); a language: family stores the detected language. This schema supports the full range of operations needed for web search: fetching the latest content for indexing, computing PageRank using link structure, and language-specific ranking — all from a single table.

CAP Theorem

The CAP theorem (Brewer, 2000; Gilbert and Lynch, 2002) states that a distributed system can provide at most two of three guarantees simultaneously:

Since network partitions are unavoidable in real systems, the practical choice is between CP (consistent, but may reject requests during a partition) and AP (available, but may return stale data during a partition). HBase chooses CP: it rejects writes to regions whose MemStore WAL cannot be confirmed, rather than risk an inconsistent state. Cassandra, by contrast, is AP: it allows writes even when some replicas are unreachable, reconciling conflicts later.

Chapter 10: Graph Analytics Redux and System Design

Pregel: Vertex-Centric Graph Processing

MapReduce’s graph algorithms suffer from the same inefficiency as any other iterative algorithm: full graph scans per iteration, with adjacency lists serialised to and from HDFS. Pregel (Malewicz et al., SOSP 2010) introduces the vertex-centric (also called “think like a vertex”) programming model.

Computation proceeds in supersteps, each comprising:

1. Each active vertex receives messages sent in the previous superstep.
2. Each active vertex runs its compute() function: reads incoming messages, updates its value, sends messages along outgoing edges, and optionally votes to halt.
3. All messages are delivered; halted vertices are reactivated if they receive a message.

Computation terminates when all vertices have voted to halt and there are no pending messages. This is the Bulk Synchronous Parallel (BSP) model: all vertices compute in parallel within a superstep, and communication occurs only at superstep boundaries.

PageRank in Pregel: each vertex stores its current rank and a list of neighbours. In each superstep, it sends rank / degree to each neighbour and updates its own rank by summing received contributions. After a fixed number of supersteps (or convergence check via aggregator), computation halts.

The advantage over MapReduce is that only active vertices (those on the current frontier, or those that received messages) need to run compute(). For BFS, only the frontier is active; the rest of the graph is idle. For PageRank, all vertices are active every superstep — the benefit is in-memory message passing rather than disk I/O.

GraphX

GraphX is Spark’s graph processing library, providing a property graph API on top of RDDs. A property graph stores arbitrary attributes on vertices and edges:

• VertexRDD: RDD[(VertexId, VD)] — vertex IDs with vertex data.
• EdgeRDD: RDD[Edge[ED]] — edges with source ID, destination ID, and edge data.
• Triplet view: RDD[EdgeTriplet[VD, ED]] — each edge augmented with source and destination vertex attributes.

The core messaging primitive is aggregateMessages:

graph.aggregateMessages[Msg](
  sendMsg = ctx => ctx.sendToDst(ctx.srcAttr),
  mergeMsg = (a, b) => a + b
)

This replaces Pregel’s compute() and combiner in a single call. sendMsg runs on each triplet; mergeMsg combines messages arriving at the same destination.

GraphX provides built-in algorithms: PageRank (both static iterations and dynamic convergence), connected components, strongly connected components, triangle counting, shortest paths, and label propagation for community detection.

System Design: Choosing the Right Tool

A key learning from this course is that no single system is optimal for all workloads. Choosing between systems requires understanding the access patterns, latency requirements, and consistency needs of the application:

Lambda and Kappa Architectures

The Lambda architecture (Nathan Marz, 2011) addresses the tension between accuracy and latency by running two parallel pipelines:

• Batch layer: reprocesses the entire historical dataset periodically to produce accurate, complete views. Slow but correct.
• Speed layer: processes only the most recent data (since the last batch run) with low latency. Fast but approximate or incomplete.
• Serving layer: merges results from batch and speed layers to answer queries.

The Lambda architecture guarantees that the batch layer will eventually correct any errors introduced by the speed layer. However, maintaining two separate codebases (one batch, one streaming) for the same business logic is operationally expensive.

The Kappa architecture (Jay Kreps, 2014) simplifies this by eliminating the batch layer: everything is a stream. Historical reprocessing is done by replaying Kafka from offset 0 with a new version of the streaming application. This works when the stream processor (e.g., Flink) is fast enough to reprocess history within an acceptable window and when the streaming API is expressive enough to handle all queries.

The Reduce-Side Join in Detail

The reduce-side join is the most general join strategy in MapReduce, but its generality comes at a cost: both full relations are shuffled across the network, incurring \(O(|R| + |S|)\) bytes of network traffic. For relations of 1 TB each, this means 2 TB of data traverses the cluster network — at 10 Gbps network bandwidth, that is over 26 minutes of network transfer time, dominating the total job time.

Performance Tuning

Achieving good performance in Spark requires attention to several dimensions:

Serialisation: Spark’s default Java serialisation is slow and produces large objects. Kryo serialisation is typically 10× faster and 5× smaller. Configure with spark.serializer = org.apache.spark.serializer.KryoSerializer.

Kryo is faster than Java serialisation because it writes only the object’s data fields (no class metadata per object), uses schema registration (the class-to-ID mapping is declared at startup rather than embedded in each serialised byte stream), and avoids Java’s reflection overhead. For a typical (String, Int) pair representing a word-count intermediate result, Java serialisation produces approximately 120 bytes (class descriptor + object header + field data); Kryo produces approximately 12 bytes (2-byte type ID + null-terminated string + 4-byte int). Across 10 billion intermediate pairs in a word count job over a 1 TB corpus, this is the difference between 1.2 TB and 120 GB of shuffle data — a 10× reduction in network and disk I/O from a single configuration change.

Partitioning: too few partitions underutilises the cluster; too many creates excessive task scheduling overhead and small shuffle files. A rule of thumb is 2–4 partitions per CPU core. Data skew (one key with far more records than others) causes stragglers — detect with the Spark UI’s stage task distribution histogram, then salt the skewed key.

Memory management: executor memory is divided between execution (shuffle, sort, aggregation) and storage (cached RDDs). The unified memory manager (Spark 1.6+) allows these regions to borrow from each other dynamically. Off-heap storage (Tungsten unsafe memory) avoids Java GC pauses for cached DataFrames.

Shuffle optimisation: the sort-based shuffle (default) spills to disk when the in-memory buffer fills. Increase spark.shuffle.memoryFraction or executor memory to reduce spills. Use reduceByKey over groupByKey to reduce shuffle data size. Enable spark.sql.shuffle.partitions tuning (default 200 is often too many for small datasets, too few for large ones).

Garbage collection: long GC pauses stall tasks and can trigger speculative execution. Use G1GC (-XX:+UseG1GC) for executors with large heaps. Monitor GC time in the Spark UI’s executor tab.

Chapter 11: Hadoop Ecosystem Deep Dive

HDFS Internals: NameNode, DataNodes, and Replication

The brief description of HDFS in Chapter 1 glossed over several engineering choices that profoundly affect correctness and performance. This chapter examines the architecture at the level required to diagnose failures and make informed deployment decisions.

The NameNode never stores block data itself. DataNodes store blocks as plain files in their local file systems, identified by block ID. On startup, each DataNode performs a block report — enumerating all block IDs present on its local disks — and sends this report to the NameNode. The NameNode cross-references block reports against its namespace to verify that every block has the required replication factor. This bootstrap protocol means that the NameNode’s in-memory block-to-DataNode mapping is rebuilt entirely from DataNode reports on every restart, rather than persisted to disk directly.

YARN: Yet Another Resource Negotiator

MapReduce 1.0 conflated resource management with job scheduling: the JobTracker managed both cluster resources and the execution of every MapReduce job. This made it impossible to run non-MapReduce workloads (Spark, Flink, MPI) on the same cluster. YARN decouples these responsibilities.

The ApplicationMaster model is the key insight: each application (a Spark job, a MapReduce job, a Flink session) brings its own scheduling logic, running in a user-space container rather than privileged RM code. A Spark AM requests containers based on Spark’s task scheduler’s needs; a MapReduce AM requests containers based on map and reduce task counts. The RM only needs to know about CPU and memory — it is agnostic to what the containers actually compute.

Hive: HiveQL to MapReduce DAG

Hive provides a SQL-like interface (HiveQL) over HDFS-stored data, compiling queries to chains of MapReduce jobs (or Tez DAGs in newer versions). Its design philosophy is “schema on read”: the schema is applied when the data is read, not when it is written. This allows raw log files, CSV files, and JSON to be queried directly by declaring an external table definition.

A HiveQL query such as:

SELECT country, COUNT(*) AS cnt
FROM page_views
WHERE date = '2025-01-01'
GROUP BY country
ORDER BY cnt DESC
LIMIT 10;

is compiled to three MapReduce jobs: (1) a map-only job that applies the WHERE predicate and projects to (country, 1) pairs; (2) a MapReduce job that shuffles by country and counts (the GROUP BY aggregation); (3) a second MapReduce job that globally sorts by count descending and returns the top 10. Each job writes its output to a temporary HDFS directory. The Hive metastore tracks table schemas and partition metadata, enabling partition pruning — if page_views is partitioned by date, job 1 reads only the date=2025-01-01 partition rather than scanning the full table.

ORC and Parquet: Columnar Storage Formats

Row-oriented formats (CSV, JSON, Avro, SequenceFile) store each record contiguously: all fields of row 1, then all fields of row 2, etc. Analytical queries typically access only a few columns out of many (e.g., SELECT SUM(revenue) FROM sales WHERE year = 2024 accesses 2 of perhaps 50 columns). With row-oriented storage, the entire row must be read from disk even though 48 of 50 columns are discarded — pure I/O waste.

Parquet and ORC are the two dominant columnar formats in the Hadoop ecosystem. Parquet (originally developed by Twitter and Cloudera) is the default format for Spark and is supported by virtually every tool in the ecosystem (Hive, Presto, Athena, BigQuery). ORC (Optimized Row Columnar, originally from Hive) achieves slightly better compression and read performance for Hive workloads due to tighter integration with the ORC reader’s vectorised execution engine. For Spark workloads, Parquet is generally preferred.

Chapter 12: Spark Internals

DAG Scheduler and Task Scheduler Separation

Spark’s execution engine is cleanly divided into two layers, each with distinct responsibilities.

The separation is important: the DAG scheduler reasons about logical dependencies between stages (which stage must complete before another can begin); the task scheduler reasons about physical placement and execution of individual tasks within a stage. This separation of concerns allows each component to be independently optimised.

RDD Dependency Types

The distinction is not merely academic. After a node failure, Spark can recover a lost narrow-dependency partition by recomputing only the upstream partition on that partition’s lineage path. For a wide dependency, the lost output partition may have been produced from records across all input partitions — recovering it requires recomputing contributions from potentially every input partition, which may require re-running the entire upstream stage.

Speculative Execution of Straggler Tasks

In large clusters, some tasks invariably run slower than others — due to network congestion, disk I/O contention, or hardware degradation. A stage cannot complete until all its tasks finish, so a single straggler can delay the entire job.

Spark Memory Management: Unified Memory Pool and Tungsten

Before Spark 1.6, memory was statically divided between execution (shuffle, sort, aggregation) and storage (cached RDDs), with a fixed boundary. If a job needed more execution memory than the allocation, tasks would spill to disk even if the storage region was empty and vice versa.

Catalyst Optimizer Pipeline

SparkSQL’s Catalyst optimizer transforms a SQL string into an executable physical plan through four sequential phases.

Chapter 13: Stream Processing Theory

Chandy-Lamport Distributed Snapshots

Exactly-once semantics in Flink rests on distributed snapshots. A naïve approach — pausing all computation, taking a snapshot, resuming — creates an unacceptable latency spike. The Chandy-Lamport algorithm (1985) takes a consistent global snapshot without stopping computation.

The key correctness property: the Chandy-Lamport snapshot captures a consistent cut — a global state where for every message recorded as received by some process, the corresponding send event is also recorded. This is exactly the condition needed for the snapshot to be a valid point to restart from.

Window Types: Formal Definitions and Worked Examples

Chapter 14: Large-Scale Graph Processing

PowerGraph: GraphLab and the Gather-Apply-Scatter Model

Pregel’s vertex-centric model places all computation in the compute() function called on each vertex. PowerGraph (Gonzalez et al., OSDI 2012) decomposes vertex computation into three explicit phases to better handle power-law graphs.

The GAS decomposition is designed for vertex-cut partitioning. When a high-degree vertex \(v\) is replicated across \(k\) machines (one “mirror” per machine that holds some of \(v\)’s edges), the Gather phase runs locally on each machine over its subset of \(v\)’s edges, producing a partial gather sum. These partial sums are sent to the “master” copy of \(v\), which combines them and runs Apply. The updated value is then sent back to all mirrors, which run Scatter locally over their edge subsets. This communication pattern is \(O(k)\) messages — proportional to the number of machines holding the vertex’s edges, not the total degree — breaking the bottleneck that Pregel faces for high-degree vertices.

Pregel SSSP on a 6-Node Graph: All Supersteps

Chapter 15: Distributed Machine Learning

Synchronous vs. Asynchronous SGD

The parameter server architecture introduced in Chapter 6 supports two update protocols with fundamentally different convergence and throughput characteristics.

AllReduce: Ring Topology and Bandwidth Optimality

Federated Learning

Chapter 16: Consistency and Coordination

Zookeeper: Distributed Coordination Service

Many distributed systems require coordination primitives: leader election (which node is the primary?), distributed locks (only one worker should process a partition), and configuration management (all nodes see the same config). Zookeeper (Hunt et al., ATC 2010) provides these as a general-purpose coordination service.

Consensus: Paxos and Raft

Consensus is the problem of getting a set of nodes to agree on a single value, despite failures. Formally: a consensus algorithm must satisfy safety (all nodes that decide, decide the same value) and liveness (all non-faulty nodes eventually decide).

Raft (Ongaro and Ousterhout, ATC 2014) repackages Paxos’s ideas into a more understandable form. Raft elects a leader explicitly; the leader receives all client writes, appends them to its log, and replicates to followers. A log entry is committed once a quorum of followers acknowledge it. On leader failure, a new leader election uses ballot numbers (Raft calls them “terms”) to prevent stale leaders from overwriting committed entries.

CRDTs: Conflict-Free Replicated Data Types

The CAP theorem (Chapter 9) forces AP systems to accept stale reads and potential write conflicts. CRDTs (Shapiro et al., 2011) are data types designed to be replicated without coordination, with a merge function that is guaranteed to converge to the same state regardless of the order or number of times replicas are merged.

Linearizability vs. Serializability

These two consistency properties are frequently confused but describe fundamentally different guarantees.

The relationship between consistency models forms a hierarchy: strict serializability > linearizability (single-object) ≈ serializability (multi-object) > snapshot isolation > read committed > eventual consistency. Distributed systems typically choose a point on this hierarchy based on their latency and availability requirements — no single point is universally optimal.