Spark: Cluster Computing with Working Sets

In one breath — the whole paper, compressed

MapReduce and Dryad made commodity clusters usable, but both are built around acyclic data flow, so every job must reload its input from disk. That is fatal for the two workloads this paper targets: iterative machine learning, which sweeps the same dataset dozens of times, and interactive analytics, where each Hive or Pig query pays tens of seconds of MapReduce startup and disk I/O. Spark's answer is the resilient distributed dataset, a read-only partitioned collection whose handle carries enough information to recompute any lost partition from data in reliable storage, so a user can pin it in memory with a cache hint and reuse it without replication or checkpointing. Around RDDs the paper adds two restricted shared variables, broadcast variables and add-only accumulators, and a Scala integration that serializes user closures and ships them to workers, including a modified Scala interpreter that makes an interactive cluster shell possible. On 20 EC2 nodes, logistic regression over 29 GB drops from 127s per iteration in Hadoop to 6s per iteration after the first, roughly 10x faster, and a 39 GB Wikipedia dump can be queried interactively in 0.5 to 1 second.

Before this paper — the world it landed in

By 2010 the MapReduce model had become the default way to compute on unreliable commodity clusters, with Dryad and Map-Reduce-Merge generalizing the shapes of data flow it supported. The bargain in all of them was the same: the user writes an acyclic graph of operators, and in exchange the system handles locality-aware scheduling, load balancing and fault tolerance without user intervention. That bargain leaked badly for anything that touched the same data more than once. Hadoop users running gradient descent had to express each iteration as an independent job that re-read the whole dataset from HDFS, and analysts running ad-hoc SQL through Pig or Hive waited tens of seconds per query because every query was a fresh MapReduce job reading from disk. The obvious alternative, distributed shared memory, had been studied for two decades but recovered from failures by checkpointing, which forced the program to roll back and cost something even when nothing failed; Twister kept static data in memory across iterations but had no fault tolerance at all and permitted only one map and one reduce function.

The problem — what was actually breaking

MapReduce, Dryad and their variants are built around an acyclic data flow model, which cannot efficiently express applications that reuse a working set of data across multiple parallel operations.
Iterative machine learning algorithms apply the same function repeatedly to the same dataset to optimize a parameter, and expressing each iteration as a separate MapReduce or Dryad job forces the data to be reloaded from disk every time, incurring a significant performance penalty.
Ad-hoc exploratory querying through SQL interfaces such as Pig and Hive incurs tens of seconds of latency per query, because each query runs as a separate MapReduce job that reads its data from disk instead of from a dataset already loaded in cluster memory.
Distributed shared memory would be general enough to fix this, but existing DSM systems achieve fault tolerance through checkpointing, which makes the program revert to a checkpoint rather than recompute only what was lost, and imposes overhead even when no node fails.
Twister, the closest prior attempt to make MapReduce iterative, keeps static data in long-lived map tasks but does not implement fault tolerance and offers only one map function and one reduce function, so a program cannot define several datasets and alternate operations over them.
No efficient, general-purpose programming language could be used interactively to process large datasets on a cluster, because closures typed line-by-line at an interpreter had no way to reach worker machines with their captured state intact.

Core ideas — the contributions, and why they work

Resilient distributed datasets

An RDD is a read-only collection of objects partitioned across a set of machines that can be rebuilt if a partition is lost. Crucially the elements need not exist in physical storage anywhere: the handle to an RDD contains enough information to compute the dataset starting from data in reliable storage, so the object is a recipe as much as it is a container. This is what makes memory safe to use as the primary residence for a working set, since losing the memory loses nothing that cannot be regenerated. The authors are explicit that RDDs are not a general shared memory abstraction, but a deliberately restricted sweet spot between expressivity on one side and scalability and reliability on the other.

Lineage instead of checkpoints

Each RDD object holds a pointer to its parent and a description of how the parent was transformed, so the chain of dataset objects is itself the recovery log. When a partition is lost, Spark re-derives just that partition by replaying its transformations against the parent, rather than reverting the whole computation to a checkpoint. Because partitions are independent, several lost partitions can be rebuilt in parallel on different nodes, and there is no overhead at all when no node fails. Lineage is cheap to capture here precisely because the programming model is restricted: with only coarse-grained deterministic transformations, a few bytes of metadata per dataset describe the entire derivation.

Persistence as a hint, not a contract

RDDs are lazy and ephemeral by default: partitions are materialized on demand when a parallel operation needs them and discarded from memory afterwards. The cache action leaves the dataset lazy but hints that it should be kept in memory after first computation, and the save action forces evaluation and writes to a distributed file system. The cache hint is not binding, and this is the point: if the cluster lacks memory for all partitions, Spark simply recomputes them when used, so programs keep working at reduced performance instead of failing. The authors call this loosely analogous to virtual memory, and frame the general goal as letting users trade off storage cost, access speed, loss probability and recomputation cost.

Two restricted shared variables

Closures passed to map, filter and reduce normally have their captured free variables copied to each worker, which is wasteful or wrong for two common patterns. A broadcast variable wraps a large read-only value, such as a lookup table or a ratings matrix, and guarantees it is copied to each worker only once rather than packaged with every closure, and it can be reused across parallel operations rather than being tied to a single job. An accumulator is a variable that workers can only add to using an associative operation and that only the driver can read, defined for any type with an add operation and a zero value. The add-only semantics are what make accumulators easy to make fault tolerant, since a re-executed task's contribution can be discarded or applied exactly once.

Closure shipping in a real language

Spark is written in Scala and exposes a functional interface in the spirit of DryadLINQ, but instead of capturing an expression tree it ships actual compiled closures. Scala closures are ordinary Java objects, so Java serialization suffices to send a computation to another machine, which is why the whole system fits in a small implementation on top of Mesos. The payoff is programmability: because Scala's for syntax desugars to foreach and accumulators support an overloaded plus-equals, the logistic regression example differs from a serial implementation in only three lines. Type inference means the programs read like ordinary Scala collection code even though every operation is a distributed job.

An interactive shell over a cluster

The Scala interpreter compiles a class per line typed by the user, containing a singleton object that holds that line's variables and runs its code in a constructor. Spark modifies this in two ways so that the same mechanism works across a cluster, letting a user define RDDs, functions, variables and classes at a prompt and use them in parallel operations. The authors claim Spark is the first system to let an efficient, general-purpose programming language be used interactively to process large datasets on a cluster. Combined with cached RDDs this changes the interaction model qualitatively: a 39 GB dataset loaded once answers subsequent full scans in under a second, comparable to working with local data.

How it works — the mechanism, concretely

Constructing and transforming RDDs

There are exactly four ways to get an RDD: read a file from a shared file system such as HDFS; parallelize a Scala collection in the driver, which slices it and sends the slices to nodes; transform an existing RDD; or change an existing RDD's persistence. The primitive transformation is flatMap, which passes each element through a user function of type A to List of B and has the same semantics as MapReduce's map; both map, of type A to B, and filter, which keeps elements matching a predicate, are expressed in terms of it. Persistence is changed by the cache action, which keeps the dataset lazy but marks it for retention in memory, and the save action, which evaluates the dataset and writes it to a distributed file system so future operations read the saved version. All of this happens in the driver program, which implements the application's high-level control flow and launches operations in parallel.

Parallel operations and the driver

Three operations trigger computation: reduce combines elements with an associative function and returns a result to the driver, collect sends all elements to the driver, and foreach passes each element through a function purely for side effects such as updating a shared variable. Nothing is materialized until one of these runs; in the text search example, errs and ones are never materialized, and when reduce is called each worker streams its input blocks, evaluates the intermediate elements, performs a local reduce, and sends only its local count to the driver. Spark at this stage supports no grouped reduce as in MapReduce, so all reduce results land in the single driver process, though local reductions do happen per node first. The authors defend this by pointing to prior work that implemented ten machine learning algorithms without parallel reduction, and plan to add a shuffle transformation later.

The RDD interface and the lineage chain

Internally every RDD implements the same three-operation interface: getPartitions returns a list of partition IDs, getIterator(partition) iterates over one partition, and getPreferredLocations(partition) reports where that partition should ideally be computed. The dataset objects form a chain that captures lineage, so the log-counting example produces HdfsTextFile, then FilteredDataset holding the contains predicate, then CachedDataset, then MappedDataset holding the mapping function, each pointing at its parent. Different RDD types differ only in how they implement those three methods: for HdfsTextFile the partitions are HDFS block IDs, the preferred locations are the block locations, and getIterator opens a stream on a block; a MappedDataset inherits its parent's partitions and locations but applies the map function in its iterator. A CachedDataset's getIterator looks for a locally cached copy of the transformed partition, and its preferred locations start as the parent's but are updated once a partition is cached on a node so future tasks prefer that node. Failure handling falls out of this design: when a node dies, its partitions are simply re-read from their parent datasets and eventually cached elsewhere.

Scheduling and shipping closures

Spark runs on Mesos, a fine-grained cluster resource manager, which lets it share a cluster and its data with Hadoop and MPI ports and which greatly reduced the implementation effort. When a parallel operation is invoked, Spark creates one task per partition and tries to send each task to one of that partition's preferred locations using delay scheduling; once launched, a task calls getIterator to start reading. Shipping a task means shipping a closure, both the closures that define a dataset and those passed to operations such as reduce, and Spark relies on Scala closures being Java objects that Java serialization can move. Scala's closure implementation was not ideal for this, because closure objects can reference outer-scope variables that the body never uses, dragging them across the network, so Spark performs a static analysis of the closure classes' bytecode to find those unused variables and nulls out the corresponding fields before serialization.

Implementing broadcast variables

Both shared variable types are implemented as classes with custom serialization formats, which is what makes them travel differently from ordinary captured variables. When a broadcast variable b is created with value v, v is written to a file in a shared file system and the serialized form of b becomes nothing more than the path to that file. When a worker first queries b's value, Spark checks a local cache and reads from the file system only on a miss, so each worker pays the transfer once no matter how many closures or operations reference it. The first implementation used HDFS for this; the paper reports that naive broadcast over HDFS or NFS made broadcast time grow linearly with node count, so the authors built an application-level multicast system and were developing a more efficient streaming broadcast.

Implementing accumulators

Each accumulator gets a unique ID at creation, and its serialized form carries only that ID plus the zero value for its type, so workers never receive the accumulated state. On a worker, a separate copy of the accumulator is created per thread using thread-local variables and reset to zero when a task begins, so concurrent tasks on the same machine never contend. After a task finishes, the worker sends the driver a message containing the updates that task made to each accumulator. The driver applies updates from each partition of each operation only once, which is the invariant that prevents double counting when a task is re-executed after a failure or when a partition is recomputed from lineage.

Interpreter integration

The stock Scala interpreter compiles a class for each line the user types, containing a singleton object with that line's variables and functions, whose constructor runs the line's code; a later reference to x compiles into a call through Line1.getInstance().x. Spark changes two things. First, the interpreter writes the classes it defines to a shared file system, from which worker nodes load them through a custom Java class loader, so workers can execute code that did not exist when the cluster started. Second, the generated code makes each line's singleton object reference previous lines' singleton objects directly instead of going through the static getInstance methods, so that a closure captures the current state of the singletons it references at serialization time; without this, an assignment such as setting x to 7 at the prompt would never propagate to the workers.

What the paper showed — measurements and proofs

Logistic regression over a 29 GB dataset on 20 m1.xlarge EC2 nodes with 4 cores each: Hadoop takes 127s per iteration because each iteration is an independent MapReduce job, while Spark takes 174s for the first iteration (likely because of Scala rather than Java) and only 6s for each subsequent iteration, letting the job run up to 10x faster.
Crashing a node mid-job in the 10-iteration logistic regression case slowed the job by 50s, or 21%, on average; the lost node's partitions were recomputed and cached in parallel on other nodes, but recovery was slower than it needed to be because a 128 MB HDFS block size left only 12 blocks per node, so recovery could not use all the cluster's cores.
Alternating least squares on 5000 movies and 15000 users on a 30-node EC2 cluster: caching the ratings matrix R in worker memory as a broadcast variable improved performance by 2.8x relative to resending R on each iteration, which otherwise dominated the job's running time.
With a naive broadcast implementation over HDFS or NFS, broadcast time grew linearly with the number of nodes and limited the scalability of the ALS job, which is why the authors implemented an application-level multicast system.
Using the modified Scala interpreter to load a 39 GB dump of Wikipedia into memory across 15 m1.xlarge EC2 machines, the first query takes roughly 35 seconds, comparable to running a Hadoop job, while subsequent queries take only 0.5 to 1 seconds even when they scan all the data.
On programmability rather than speed, the paper reports that the Spark logistic regression program differs from a serial version of the same algorithm in only three lines, thanks to accumulators plus Scala's for syntax desugaring into a parallel foreach.

Limits and trade-offs — conceded and discovered

Conceded in the paper: Spark supports no grouped reduce as in MapReduce, so all reduce results are collected at the single driver process; group-bys and joins are impossible until the shuffle transformation listed as future work in the discussion is built.
Conceded in the paper: the cache action is only a hint, so a working set larger than cluster memory silently degrades into recomputation, and the only persistence levels offered are in-memory caching and saving to a distributed file system, with in-memory replication and a general storage-cost versus reconstruction-cost knob left as future work.
Conceded in the paper: fault recovery is not free or instant, costing 21% of the job in the measured failure experiment and being limited by coarse 128 MB blocks, and Spark's first iteration is actually slower than Hadoop's per-iteration time (174s versus 127s), so the win exists only when data is reused.
Conceded in the paper: RDDs are explicitly not a general shared memory abstraction, they are read-only with no fine-grained writes, and the authors list formally characterizing the properties of RDDs and their suitability for various workloads as an open item; the implementation is described throughout as a prototype at an early stage.
Exposed by later work: with no checkpointing, lineage chains grow without bound in long iterative jobs and recovery cost grows with them, and the coarse treatment of dependencies here gives the scheduler nothing to reason about; the 2012 NSDI RDD paper added narrow versus wide dependencies, stage-based scheduling and checkpointing to fix exactly these gaps.

What it became — the systems that inherited it

This four-and-a-half-page workshop paper is the seed of Apache Spark, which went on to displace Hadoop MapReduce as the default engine for large-scale batch processing. Its own direct sequel, the NSDI 2012 paper on resilient distributed datasets, kept the lineage idea intact and hardened it with narrow versus wide dependency classification, stage-based DAG scheduling, and checkpointing for long lineage chains. The shuffle transformation promised in the discussion section arrived and enabled group-bys and joins, which in turn made Shark and then Spark SQL, DataFrames and the Catalyst optimizer possible; the interactive interpreter sketched here became the spark-shell and the notebook-driven workflow that Databricks commercialized. The other abstractions propagated too: broadcast variables and accumulators survive essentially unchanged in modern Spark, and the same in-memory reuse argument was carried into MLlib for iterative learning, GraphX for graph algorithms, and Spark Streaming, whose discretized streams are just RDDs over time windows and inherit lineage recovery instead of upstream backup. More broadly, the core claim that a dataset handle carrying enough information to reconstruct itself is a better fault-tolerance primitive than replication or checkpointing shows up in later dataflow systems such as Flink and Dask, and in the lakehouse lineage of Delta Lake, where deterministic recomputation from durable inputs remains the recovery story.

In the paper’s words — verbatim

“RDDs achieve fault tolerance through a notion of lineage: if a partition of an RDD is lost, the RDD has enough information about how it was derived from other RDDs to be able to rebuild just that partition.”

§1

“We note that our cache action is only a hint: if there is not enough memory in the cluster to cache all partitions of a dataset, Spark will recompute them when they are used.”

§2.1

“Although RDDs are not a general shared memory abstraction, they represent a sweet-spot between expressivity on the one hand and scalability and reliability on the other hand, and we have found them well-suited for a variety of applications.”

§1

Vocabulary — as this paper uses it

Resilient distributed dataset (RDD): A read-only collection of objects partitioned across a set of machines that can be rebuilt if a partition is lost. Its elements need not exist in physical storage; the handle carries enough information to compute the dataset from data in reliable storage.
Lineage: The chain of dataset objects recording, for each RDD, a pointer to its parent and how the parent was transformed. Spark replays this chain to recompute just a lost partition, instead of checkpointing and rolling back.
Working set: A body of data that an application reuses across multiple parallel operations, as in iterative machine learning or repeated interactive queries. It is the workload class that acyclic data flow systems handle badly and that Spark is designed for.
Parallel operation: An action that triggers computation on an RDD by shipping a closure to workers: reduce, which combines elements with an associative function and returns to the driver, collect, which sends all elements to the driver, and foreach, which runs a function for its side effects.
Driver program: The user's main program, which implements the application's high-level control flow, defines RDDs and shared variables, and launches parallel operations on the cluster. All reduce and collect results return to it.
Broadcast variable: A wrapper around a large read-only value that ensures the value is copied to each worker only once instead of being packaged with every closure. Its serialized form is just a path to a file in a shared file system, and it is reusable across parallel operations.
Accumulator: A shared variable that workers can only add to using an associative operation and that only the driver can read, defined for any type with an add operation and a zero value. Its add-only semantics make it easy to make fault tolerant, and the driver applies each partition's updates only once.
cache action: A persistence change that leaves an RDD lazy but hints that its partitions should be kept in memory after first computation. It is only a hint: if cluster memory is insufficient, Spark recomputes the partitions when they are used.
Preferred locations: The per-partition placement hints returned by getPreferredLocations and used by delay scheduling to send each task where its data lives. For a cached dataset they start as the parent's locations and are updated once a partition is cached on a node.

On the timeline — where this sits in the story

Year2010

MilestoneApache Spark

Problem solvedMapReduce was inefficient for iterative machine learning, interactive exploration, and multi-stage pipelines.

How it solved itSpark introduced resilient distributed datasets and memory-oriented cluster computation.

Why it matteredBecame a dominant general-purpose big-data engine for ETL, ML, streaming, and SQL workloads.

DDIACh. 11: Batch Processing

View on the timeline