study › DDIA › Chapter 11 · Batch Processing

Materializing Between Stages Costs a 3×-Replicated Write and Blocks the First Output

The same three-stage workflow (map → group-by → aggregate) over 100,000 records, two engines. The MapReduce-style engine writes each stage's full output to a replicated DFS — 2,918,670 bytes of intermediate state — and its first final record appears only after the entire input has flowed through all three stages (100,000 records consumed). The dataflow pipeline writes 0 intermediate bytes and emits its first record after just 101 — a 990× gap.

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.

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.
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.
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".
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 prerequisite 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. Together they are why the intermediate is larger than the final result and why the first record waits for the last input.

Environment these numbers came from Deterministic, so the numbers reproduce exactly. Captured on 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 and byte counts are deterministic; time-to-first-output is a records-consumed proxy, not wall-clock — it counts source records pulled before the first final record is emitted, a property of the dataflow, not the clock. Fixed workload: 100,000 records, 1,000 keys × 100 records each, 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.

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?

Do the two engines agree, and on how much input?

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?

How many intermediate bytes does each engine write?

(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?

How many input records before the first final aggregate?

(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

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?
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.

Sources: DDIA 2e, Ch. 11, §"Materialization of Intermediate State", §"Dataflow Engines" · Dean & Ghemawat, "MapReduce" (2004) · Ghemawat et al., "The Google File System" (2003).