Resilient Distributed Datasets

"They asked where I came from. I did not have a backup, but I had directions: take the source, map it, filter it, and you will get me again."

A Partition That Was Recomputed, Not Restored

Section 7.1 told the story of why Spark followed MapReduce: the same partition-and-recombine shape, but with the intermediate results kept in memory and reused across many operations instead of written to disk after every step. To keep many operations cheap and still survive the failures that Section 2.4 warns are routine at cluster scale, Spark needed an abstraction that was cheap to hold in memory, cheap to recombine, and cheap to recover. That abstraction is the RDD. Everything higher in Spark, the DataFrames and Spark SQL of Section 7.3 included, compiles down to operations on RDDs, so understanding this layer is understanding what Spark actually does when you press run.

An RDD has three defining properties, and each one earns its place. It is partitioned: the records are split into chunks, one chunk per unit of parallel work, so a cluster of machines can process them at once. It is immutable: a transformation never edits an RDD in place, it produces a new RDD, which means an existing RDD can be read concurrently and recomputed at any time without fear that someone changed it underneath you. And it carries its lineage: a record of the parent RDDs and the transformation that derived it from them. Immutability plus lineage is the combination that makes the rest of this section possible; hold onto it.

1. Lineage: The Recipe Instead of the Replica Beginner

The classic way to survive a lost machine is replication: keep two or three copies of every block of data on different nodes, so when one dies the others still have it. That is how the distributed filesystem under MapReduce protects its data, and it works, but it is expensive. Holding three copies of a multi-terabyte intermediate dataset in cluster memory triples the memory you must rent, and for data that is being processed and thrown away within minutes, paying to store three copies of it is a poor trade.

RDDs take the other route. For each RDD, Spark records its lineage: the list of parent RDDs it was built from and the transformation that produced it, such as "this RDD is the result of applying map(square) to that RDD." Chaining these records gives a directed acyclic graph (DAG) of derivations that traces every RDD back to data on stable storage. When a partition is lost because the machine holding it crashed, Spark does not look for a replica; it walks the lineage and recomputes just that partition from its parents. Because the inputs were immutable and the transformations are deterministic functions, the recomputed partition is identical to the one that was lost, which is exactly what Section 2.4 calls deterministic re-execution as a recovery strategy.

The cost profile is the opposite of replication. Replication pays continuously, in extra memory and bandwidth for every copy, whether or not a failure ever happens. Lineage pays almost nothing in the common case, just the few bytes to store each derivation step, and pays the recomputation cost only when a partition is actually lost, and only for the partitions that were lost. For a long chain of narrow transformations, recomputing one partition re-runs that chain on one partition's worth of input, which is a small, local job. The catch arrives with wide dependencies, and that is the next idea.

2. Narrow and Wide Dependencies Intermediate

Not all lineage edges are equal. The crucial distinction is how many input partitions an output partition depends on. In a narrow dependency, each partition of the child RDD depends on at most one partition of each parent. Transformations like map and filter are narrow: to produce output partition $i$, you read input partition $i$, apply the function record by record, and you are done. No data leaves the machine, and the whole chain of narrow transformations can run as one pipelined pass over each partition.

In a wide dependency, a single output partition can depend on records from many input partitions. Transformations like groupByKey, reduceByKey, and join are wide: to gather all values for a given key into one place, you must pull records carrying that key out of every input partition and route them to the partition that will hold the group. That routing of records across the network, sorting by key and sending each record to its destination partition, is the shuffle, the same expensive all-to-all movement introduced for MapReduce in Section 6.2. A wide dependency forces a shuffle, and a shuffle forces a stage boundary: Spark cannot pipeline through it, because the output side cannot start until the input side has produced and shipped all the records the shuffle will move.

If $p_{\text{in}}$ is the number of input partitions and $p_{\text{out}}$ the number of output partitions, a narrow dependency wires each output partition to one input partition, so the number of dependency edges is $O(p_{\text{out}})$, while a wide dependency can wire every output partition to every input partition, up to $O(p_{\text{in}} \cdot p_{\text{out}})$ edges of data movement. That blow-up is why the shuffle dominates the cost of most Spark jobs, a theme that returns in full in Section 7.3 and the partitioning discussions later in this chapter. Figure 7.2.1 draws both kinds of edge and the stage boundary that the wide one creates.

Narrow and wide dependencies also change the price of recovery. Re-running a chain of narrow transformations to rebuild one lost partition is cheap and local, as Section 1 promised. Re-running a partition that sits downstream of a wide dependency is not: its inputs came from a shuffle, so recomputing it may require recomputing the shuffle, which may require re-reading many parent partitions. This is why Spark checkpoints or persists across wide dependencies in long jobs, cutting the lineage so that recovery does not have to replay an expensive shuffle. Lineage is cheap insurance until a shuffle sits in the chain; then it has a real premium.

3. An RDD You Can Hold in Your Hand Intermediate

The ideas above are easy to state and easy to misjudge, so we make them concrete. The code below is a complete, dependency-free miniature of the RDD model in a few dozen lines of plain Python. Each transformation builds a new immutable object, records its parent and whether the dependency is narrow or wide, and stores a closure that knows how to compute any one partition on demand. Nothing is materialized until a partition is asked for, and a "lost" partition is recovered by simply asking for it again, which re-runs the closure against the immutable parents. This is the same mechanism Spark uses, stripped of the networking.

NARROW, WIDE = "narrow", "wide"

class MiniRDD:
    def __init__(self, parents, op_name, dep, compute_partition, num_partitions):
        self.parents = parents              # the RDDs this one was derived from
        self.op_name = op_name              # human-readable transformation name
        self.dep = dep                      # NARROW or WIDE
        self._compute = compute_partition   # (index) -> list of records
        self.num_partitions = num_partitions

    def map(self, f):                       # narrow: per-record, per-partition
        return MiniRDD([self], f"map({f.__name__})", NARROW,
                       lambda i: [f(x) for x in self.partition(i)],
                       self.num_partitions)

    def filter(self, pred):                 # narrow: keep some records, stay local
        return MiniRDD([self], f"filter({pred.__name__})", NARROW,
                       lambda i: [x for x in self.partition(i) if pred(x)],
                       self.num_partitions)

    def group_by_key(self):                 # WIDE: each output reads EVERY input
        def compute(i):
            buckets = {}
            for src in range(self.num_partitions):       # read all parent partitions
                for k, v in self.partition(src):
                    if hash(k) % self.num_partitions == i:   # route key to its bucket
                        buckets.setdefault(k, []).append(v)
            return sorted(buckets.items())
        return MiniRDD([self], "group_by_key", WIDE, compute, self.num_partitions)

    def partition(self, i):                 # compute (or RE-compute) one partition
        return self._compute(i)

    def lineage(self):                      # walk parents back to the source
        chain, node = [], self
        while node.parents:
            chain.append((node.op_name, node.dep)); node = node.parents[0]
        chain.append((node.op_name, node.dep))
        return list(reversed(chain))

def from_collection(records, num_partitions):
    shards = [records[i::num_partitions] for i in range(num_partitions)]
    return MiniRDD([], "source", NARROW, lambda i: list(shards[i]), num_partitions)

def even(x):   return x % 2 == 0
def square(x): return x * x
def tag(x):    return (x % 3, x)            # key by remainder, value is the number

base = from_collection(list(range(12)), num_partitions=3)
narrow_chain = base.map(square).filter(even)        # map then filter: all narrow
wide_chain   = base.map(tag).group_by_key()         # group_by_key: wide

print("=== Lineage of the narrow chain ===")
for op, dep in narrow_chain.lineage():
    print(f"  {op:18s} -> {dep}")

print("\n=== Lineage of the wide chain (note the stage boundary) ===")
for op, dep in wide_chain.lineage():
    boundary = "   <-- shuffle / stage boundary" if dep == WIDE else ""
    print(f"  {op:18s} -> {dep}{boundary}")

print("\n=== Normal result of narrow_chain, partition 1 ===")
print(" ", narrow_chain.partition(1))

# Simulate LOSING partition 1 (a worker crashed). No replica exists; we just
# recompute it from the lineage by asking for it again.
print("\n=== Partition 1 lost. Recomputing from lineage (no replica needed) ===")
recovered = narrow_chain.partition(1)               # re-derive from parents on demand
print("  recovered partition 1 :", recovered)

print("\n=== Determinism: recompute equals the original ===")
print("  match:", recovered == narrow_chain.partition(1))

all_edges = narrow_chain.lineage() + wide_chain.lineage()
nw = sum(1 for _, d in all_edges if d == NARROW)
wd = sum(1 for _, d in all_edges if d == WIDE)
print(f"\nclassified edges: {nw} narrow, {wd} wide "
      f"({wd} shuffle => {wd} extra stage boundary)")

=== Lineage of the narrow chain ===
  source             -> narrow
  map(square)        -> narrow
  filter(even)       -> narrow

=== Lineage of the wide chain (note the stage boundary) ===
  source             -> narrow
  map(tag)           -> narrow
  group_by_key       -> wide   <-- shuffle / stage boundary

=== Normal result of narrow_chain, partition 1 ===
  [16, 100]

=== Partition 1 lost. Recomputing from lineage (no replica needed) ===
  recovered partition 1 : [16, 100]

=== Determinism: recompute equals the original ===
  match: True

classified edges: 5 narrow, 1 wide (1 shuffle => 1 extra stage boundary)

Three things in Output 7.2.1 are worth pausing on. First, the recovered partition equals the original exactly, with no replica anywhere in the program; the only thing that survived the loss was the lineage. Second, recomputing partition 1 of the narrow chain touched only partition 1's parents, never partitions 0 or 2, which is the locality that makes narrow recovery cheap. Third, the dependency classifier counted exactly one wide edge across both chains, and that single wide edge is the only place a real Spark scheduler would insert a stage boundary and pay for a shuffle. The miniature reproduces the model's behavior, not just its vocabulary.

# Run with a Spark session already created as `sc` (the SparkContext).
base = sc.parallelize(range(12), numSlices=3)       # an RDD with 3 partitions

narrow = base.map(lambda x: x * x).filter(lambda x: x % 2 == 0)   # narrow, pipelined
wide   = base.map(lambda x: (x % 3, x)).groupByKey()             # wide -> a shuffle

print(narrow.collect())            # [0, 4, 16, 36, 64, 100]
print(wide.toDebugString().decode())   # prints the lineage DAG and the stage split
# If a worker dies mid-job, Spark recomputes only the lost partitions from lineage;
# you never write recovery code.

4. Why Immutability Buys Determinism, and Determinism Buys Recovery Advanced

It is worth making the logical chain explicit, because it is the load-bearing argument of the whole RDD model. Lineage-based recovery rests on one claim: recomputing a partition from its parents yields the same partition that was lost. That claim is true only if recomputation is deterministic, and recomputation is deterministic only if two conditions both hold. The transformation must be a pure function of its inputs, the same inputs always producing the same outputs, and the inputs must not have changed since the first computation. Immutability guarantees the second condition outright: an RDD, once built, is never modified, so its partitions are the same data on the replay as on the original run. Purity of the common transformations (map, filter, and friends apply a side-effect-free function record by record) guarantees the first.

Determinism does more than make recovery correct; it makes the entire computation reproducible. Run the same RDD program twice on the same input and you get the same result, partition by partition, which is what lets Spark speculatively re-run a slow straggler task on another machine and safely keep whichever copy finishes first, a straggler-mitigation idea that Section 2.4 frames in general terms. The places where this breaks are precisely the places to be careful: a transformation that reads the wall clock, draws a random number from an unseeded generator, or depends on the arrival order of shuffled records is not a pure function, and an RDD built from it cannot be safely recomputed. Spark's reliability is, at bottom, a discipline of keeping transformations deterministic.

The narrow/wide distinction has a determinism wrinkle too. Within a partition, narrow transformations preserve record order, so their output is deterministic down to the ordering. A shuffle, by contrast, gathers records from many partitions whose network arrival order is not fixed, so unless you sort, the order of records inside a post-shuffle partition can vary run to run even though the set of records does not. For correctness this rarely matters, since most aggregations are order-insensitive, but it is the reason a transformation that secretly depends on within-partition order can pass every test and then fail mysteriously after a shuffle. Determinism of the multiset is guaranteed; determinism of the order is not, across a wide dependency.

We now have the RDD in full: a partitioned, immutable collection that records its lineage, recovers lost partitions by deterministic recomputation rather than replication, and splits its work at exactly the wide dependencies where a shuffle, and a stage boundary, become unavoidable. This is the engine under everything Spark does. The next section climbs one level up to the DataFrame and Spark SQL, where the same RDD machinery hides beneath a structured, optimizable interface that a query planner can reorganize on your behalf; that ascent begins in Section 7.3.