# DDIA Ch. 11: materializing between stages costs a 3×-replicated write and blocks the first output

## Concept

Run the SAME three-stage batch workflow — map → group-by → aggregate
— over ONE dataset of 100,000 records, TWO ways. The **MapReduce-style**
engine writes each stage's *full* output to a distributed filesystem before the
next stage may read it: we model the DFS honestly (write real bytes, `fsync`,
count 3 replica copies), so the intermediate data between stages is written
**2,918,670 bytes** — and because every stage's file must be complete before
the next opens it, the first *final* record cannot appear until the entire input
has passed through all three stages: **100,000 of 100,000** records consumed
first. The **dataflow-style** engine chains the same three stages as generators:
nothing is written between stages (**0 bytes**), and the first final record is
emitted after only **101** input records — the first group plus the one
record that closes it. Same workflow, same 1,000 aggregates; the only difference
is whether intermediate state is materialized to disk or streamed in memory.

The prediction that fails: "writing between stages is a minor overhead." It is a
full, 3×-replicated copy of the data at *every* stage boundary, and it
imposes a barrier that forbids the stages from overlapping at all.

## Provenance

*Designing Data-Intensive Applications*, 2nd ed. (Kleppmann & Riccomini, 2025),
Chapter 11 — *Batch Processing*, §"Materialization of Intermediate
State" and §"Dataflow Engines":

> Writing out this intermediate state to files is called *materialization* [...]
> a MapReduce job can only start when all tasks in the preceding jobs have
> completed, whereas processes connected by a Unix pipe are started at the same
> time.

The book contrasts MapReduce, which materializes every operator's whole output to
replicated storage before the next reads it, against dataflow engines (Spark,
Flink, Tez) that hand records operator-to-operator without landing them on disk.
Here you measure both halves of that trade: the replicated intermediate bytes,
and the barrier that delays the first output.

## Prerequisites

The result turns on what a stage boundary *is* in each engine. Each line notes
where it shows up in the run.

1. **MapReduce stage boundaries & the distributed filesystem** — a MapReduce
   job's output is written to a replicated DFS (HDFS/GFS); the next job reads it
   back. Stages communicate only through files. It shows up as `materialize()`
   writing each stage's output before the next stage reads it. *Free resource:*
   Dean & Ghemawat, "MapReduce: Simplified Data Processing on Large Clusters"
   (2004), §3 Implementation.
2. **Replication factor** — the DFS keeps N durable copies of every block
   (N=3 by default) so a lost disk loses no data. It shows up as the
   `* REPLICATION` in the byte count: every intermediate byte is written three
   times. *Free resource:* Ghemawat et al., "The Google File System" (2003),
   §2.4 Replica Placement.
3. **Materialization vs. pipelining** — *materialize* = write the whole
   output, then let the next stage read it (durable, restartable, serial);
   *pipeline* = stream each record straight to the next operator (in-memory,
   overlapping, must recompute on failure). It shows up as the two functions
   `run_mapreduce` vs. `run_pipeline`. *Free resource:* DDIA 2e §"Dataflow
   Engines".
4. **Operator fusion / generators** — chaining lazy iterators so a record is
   pulled through every stage on demand, with no buffer between them, is exactly
   how a dataflow engine fuses operators into a pipeline. It shows up as the
   `(parse(raw) for raw in source)` → `streaming_group(...)` →
   `((k, sum(vs)) ...)` chain. *Free resource:* Python docs, "Generators" and
   PEP 289 (generator expressions).

The one that carries the result: **#2, the replication factor** turns "a write
between stages" into "three full durable copies at every boundary" — and
**#3's barrier** turns that serial write into a wall the first output cannot cross
until the whole input has.

## Environment

Deterministic, so the numbers reproduce exactly:

- macOS 26.5.2 (Darwin 25.5.0), arm64
- Python 3.15.0a8, standard library only — no Docker, no deps
- The replication factor and stage boundaries are modeled explicitly; byte counts
  are deterministic. Time-to-first-output is a **records-consumed proxy**, not
  wall-clock — it counts how many source records are pulled before the first
  final record is emitted, which is a property of the dataflow, not of the clock.
- Fixed workload: 100,000 records, 1,000 keys × 100 records each, laid out
  already grouped by key; DFS replication factor = 3.

## Mental model

Same workflow, same data — only what happens *between* stages differs.

- **MapReduce (materialize)** — each operator writes its whole output to
  replicated storage before the next operator reads it. Durable and restartable
  (a failed stage just re-reads its input file), but serial and write-heavy: the
  next stage cannot begin until the previous stage's file is complete, so stages
  never overlap and every byte between them is written 3×.
- **Dataflow engine (pipeline)** — records stream operator-to-operator in
  memory. Fast and pipelined (the first output can appear before the last input is
  read) and it writes nothing between stages — but the intermediate state is
  only in RAM, so a failure means recomputing from a checkpoint or the source.

Both engines run the identical map → group-by → aggregate over the
identical dataset and must produce the identical 1,000 aggregates. The whole thing
is one script, `code/materialization_vs_pipelining.py`.

## Setup

```
cd study/ddia/ch11/code
python3 materialization_vs_pipelining.py
```

Do **not** read the output yet — make each prediction first.

## Steps

### Step 1 · both engines, one answer

**Predict.** Two different engines run the same map → group-by →
aggregate over the same 100,000 records. Should their 1,000 aggregates be
identical, and roughly how big is the *input* to each?

**Observe.**

```
one workflow (map -> group-by -> aggregate) over 100,000 records, two engines
    1,000 keys x 100 records each, DFS replication factor = 3

both engines produce the identical 1,000 aggregates (e.g. key 0 -> 4050, key 999 -> 4950).
```

Identical results — as they must be, it is the same computation. The 100,000
records are the shared input. The surprise isn't the answer; it's what each engine
*spends* to get there.

### Step 2 · intermediate bytes written to the DFS

**Predict.** The map output and the group output each get materialized to the DFS
before the next stage reads them, and the DFS keeps 3 replicas. The pipeline
writes nothing between stages. Is the materialized intermediate a "minor
overhead," and what does the pipeline write — how many bytes each?

**Observe.**

```
(1) intermediate bytes written to the DFS between stages
    MapReduce (materialize map + group, each x3): 2,918,670 bytes (2850.3 KiB)
    dataflow pipeline (generators, nothing written):          0 bytes (0.0 KiB)
    ratio: MapReduce writes 2,918,670 bytes of intermediate; the pipeline writes 0.
```

**2,918,670 bytes vs. 0.** Not a minor overhead: the map output and the group
output are each a full copy of the data, and each is written 3× for
replication. The pipeline never lands intermediate state on disk at all —
those bytes live in memory and are gone the instant the next operator consumes
them.

### Step 3 · time to first output

**Predict.** How many of the 100,000 input records must each engine consume before
it can emit its *first* final aggregate? Same number for both, or different?

**Observe.**

```
(2) time-to-first-output  (source records pulled before the FIRST final record)
    MapReduce:         100,000 of 100,000 records  (the ENTIRE input, at every stage barrier)
    dataflow pipeline: 101 of 100,000 records  (just the first group + the record that closed it)
    ratio: MapReduce consumes 990x more input before it can emit anything.
```

**100,000 vs. 101.** MapReduce's stage 3 (aggregate) cannot start until stage 2's
file is complete, which cannot start until stage 1 has consumed *every* input
record — so the first final record only exists after the whole input has
flowed through all three stages. The pipeline pulls records through map →
group → aggregate on demand: once the first group's 100 records are in and the
101st record (a new key) closes that group, the first aggregate falls out
— 990× sooner.

## What you should see

- Both engines produce the **identical 1,000 aggregates** (key 0 → 4050, key
  999 → 4950).
- MapReduce writes **2,918,670 bytes** of replicated intermediate state; the
  pipeline writes **0**.
- MapReduce consumes **100,000 of 100,000** input records before its first final
  output; the pipeline consumes **101** — a **990×** gap.

## Why

A stage boundary in MapReduce is a *file* in a replicated distributed filesystem,
and that single design choice produces both numbers from first principles.

**The 3×.** The map stage emits one (key, value) pair per input record; the
group stage emits those same values, regrouped. Neither shrinks the data —
the intermediate volume between stages is on the order of the input itself. The DFS
then writes each block to `REPLICATION = 3` machines so a lost disk loses no data.
So the intermediate bytes are `3 * (map_output + group_output)`: two full stage
outputs, each tripled. That is where 2,918,670 comes from — not a rounding
error on the "real" work, but a quantity *larger* than the final result, paid at
every boundary. Materialization buys fault tolerance with it: because each stage's
input sits durably on disk, a crashed stage just re-reads its file and re-runs,
with no need to recompute the stages before it.

**The barrier.** Materialization also serializes the stages. Stage 2 opens stage
1's file, so it may not start until that file is *complete*; stage 3 waits on stage
2 the same way. A file is complete only after its producer has consumed all of its
input. Chain three such stages and the first final record cannot exist until the
entire input has been read, mapped, written, re-read, grouped, written, re-read,
and aggregated — hence 100,000 records consumed before the first output. The
pipeline removes the files: each operator is a generator that yields to the next,
so a record flows straight through, and the first aggregate emerges as soon as the
first group is closed (record 101). No file means no barrier; nothing forces the
engine to see the whole input before producing part of the answer.

Pipelining trades exactly what materialization bought. Its intermediate state is in
RAM, so there is nothing to re-read after a crash: a failure forces recomputation
from the last checkpoint or the source. Durability-by-materialization vs.
latency-and-throughput-by-pipelining is the whole trade — the 3× write
and the barrier are the price of the first, and their absence is the price of the
second.

### The boundary — when each engine wins

Materialization wins when stages are **expensive** and **failures are common**
(long multi-hour jobs on thousands of machines): re-reading one stage's file beats
recomputing the whole lineage, and the write cost is amortized over hours of
compute. Pipelining wins for **latency** and when **re-running is cheap** (short
jobs, interactive queries, streaming): the first output appears immediately and no
disk is touched between operators. Real engines are hybrids — Spark pipelines
within a stage but materializes (spills/shuffles) at stage boundaries; Flink
pipelines and checkpoints periodically. The choice is per-boundary, not
per-engine.

### Go deeper

1. **Spark's lineage-based recompute.** Spark doesn't replicate intermediate state
   3× like the DFS here — it records the *lineage* (the operations that
   produced a partition) and recomputes a lost partition from its parents. Read
   that as: it moves the fault-tolerance cost off the write path (no 3×) and
   onto the recovery path (recompute on failure). When is recompute cheaper than a
   replicated write, and when does Spark `persist()`/replicate anyway?
2. **Why a sort or a barrier still forces materialization.** This exercise streams
   the group-by *because the input is already grouped by key* — a group is
   complete the moment a new key arrives. Shuffle the input, or ask for a global
   sort or a full-input aggregate (a median, a top-K over everything), and the
   operator can't emit its first output until it has seen the *last* record. That
   re-introduces a barrier even inside a dataflow engine, and the engine
   materializes at that point. Change the dataset to random key order and watch the
   streaming group-by produce wrong (split) groups — the reason real shuffles
   sort first, and pay to materialize.