Resilient Distributed Datasets

1 . Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing Matei Zaharia, Mosharaf Chowdhury, Tathagata Das, Ankur Dave, Justin Ma, Murphy McCauley, Michael J. Franklin, Scott Shenker, Ion Stoica University of California, Berkeley Abstract tion, which can dominate application execution times. We present Resilient Distributed Datasets (RDDs), a dis- Recognizing this problem, researchers have developed tributed memory abstraction that lets programmers per- specialized frameworks for some applications that re- form in-memory computations on large clusters in a quire data reuse. For example, Pregel [22] is a system for fault-tolerant manner. RDDs are motivated by two types iterative graph computations that keeps intermediate data of applications that current computing frameworks han- in memory, while HaLoop [7] offers an iterative MapRe- dle inefficiently: iterative algorithms and interactive data duce interface. However, these frameworks only support mining tools. In both cases, keeping data in memory specific computation patterns (e.g., looping a series of can improve performance by an order of magnitude. MapReduce steps), and perform data sharing implicitly To achieve fault tolerance efficiently, RDDs provide a for these patterns. They do not provide abstractions for restricted form of shared memory, based on coarse- more general reuse, e.g., to let a user load several datasets grained transformations rather than fine-grained updates into memory and run ad-hoc queries across them. to shared state. However, we show that RDDs are expres- In this paper, we propose a new abstraction called re- sive enough to capture a wide class of computations, in- silient distributed datasets (RDDs) that enables efficient cluding recent specialized programming models for iter- data reuse in a broad range of applications. RDDs are ative jobs, such as Pregel, and new applications that these fault-tolerant, parallel data structures that let users ex- models do not capture. We have implemented RDDs in a plicitly persist intermediate results in memory, control system called Spark, which we evaluate through a variety their partitioning to optimize data placement, and ma- of user applications and benchmarks. nipulate them using a rich set of operators. The main challenge in designing RDDs is defining a 1 Introduction programming interface that can provide fault tolerance Cluster computing frameworks like MapReduce [10] and efficiently. Existing abstractions for in-memory storage Dryad [19] have been widely adopted for large-scale data on clusters, such as distributed shared memory [24], key- analytics. These systems let users write parallel compu- value stores [25], databases, and Piccolo [27], offer an tations using a set of high-level operators, without having interface based on fine-grained updates to mutable state to worry about work distribution and fault tolerance. (e.g., cells in a table). With this interface, the only ways to provide fault tolerance are to replicate the data across Although current frameworks provide numerous ab- machines or to log updates across machines. Both ap- stractions for accessing a cluster’s computational re- proaches are expensive for data-intensive workloads, as sources, they lack abstractions for leveraging distributed they require copying large amounts of data over the clus- memory. This makes them inefficient for an important ter network, whose bandwidth is far lower than that of class of emerging applications: those that reuse interme- RAM, and they incur substantial storage overhead. diate results across multiple computations. Data reuse is common in many iterative machine learning and graph In contrast to these systems, RDDs provide an inter- algorithms, including PageRank, K-means clustering, face based on coarse-grained transformations (e.g., map, and logistic regression. Another compelling use case is filter and join) that apply the same operation to many interactive data mining, where a user runs multiple ad- data items. This allows them to efficiently provide fault hoc queries on the same subset of the data. Unfortu- tolerance by logging the transformations used to build a nately, in most current frameworks, the only way to reuse dataset (its lineage) rather than the actual data.1 If a parti- data between computations (e.g., between two MapRe- tion of an RDD is lost, the RDD has enough information duce jobs) is to write it to an external stable storage sys- about how it was derived from other RDDs to recompute tem, e.g., a distributed file system. This incurs substantial 1 Checkpointing the data in some RDDs may be useful when a lin- overheads due to data replication, disk I/O, and serializa- eage chain grows large, however, and we discuss how to do it in §5.4. 

2 .just that partition. Thus, lost data can be recovered, often differentiate them from other operations on RDDs. Ex- quite quickly, without requiring costly replication. amples of transformations include map, filter, and join.2 Although an interface based on coarse-grained trans- RDDs do not need to be materialized at all times. In- formations may at first seem limited, RDDs are a good stead, an RDD has enough information about how it was fit for many parallel applications, because these appli- derived from other datasets (its lineage) to compute its cations naturally apply the same operation to multiple partitions from data in stable storage. This is a power- data items. Indeed, we show that RDDs can efficiently ful property: in essence, a program cannot reference an express many cluster programming models that have so RDD that it cannot reconstruct after a failure. far been proposed as separate systems, including MapRe- Finally, users can control two other aspects of RDDs: duce, DryadLINQ, SQL, Pregel and HaLoop, as well as persistence and partitioning. Users can indicate which new applications that these systems do not capture, like RDDs they will reuse and choose a storage strategy for interactive data mining. The ability of RDDs to accom- them (e.g., in-memory storage). They can also ask that modate computing needs that were previously met only an RDD’s elements be partitioned across machines based by introducing new frameworks is, we believe, the most on a key in each record. This is useful for placement op- credible evidence of the power of the RDD abstraction. timizations, such as ensuring that two datasets that will We have implemented RDDs in a system called Spark, be joined together are hash-partitioned in the same way. which is being used for research and production applica- 2.2 Spark Programming Interface tions at UC Berkeley and several companies. Spark pro- vides a convenient language-integrated programming in- Spark exposes RDDs through a language-integrated API terface similar to DryadLINQ [31] in the Scala program- similar to DryadLINQ [31] and FlumeJava [8], where ming language [2]. In addition, Spark can be used inter- each dataset is represented as an object and transforma- actively to query big datasets from the Scala interpreter. tions are invoked using methods on these objects. We believe that Spark is the first system that allows a Programmers start by defining one or more RDDs general-purpose programming language to be used at in- through transformations on data in stable storage teractive speeds for in-memory data mining on clusters. (e.g., map and filter). They can then use these RDDs in We evaluate RDDs and Spark through both mi- actions, which are operations that return a value to the crobenchmarks and measurements of user applications. application or export data to a storage system. Examples We show that Spark is up to 20× faster than Hadoop for of actions include count (which returns the number of iterative applications, speeds up a real-world data analyt- elements in the dataset), collect (which returns the ele- ics report by 40×, and can be used interactively to scan a ments themselves), and save (which outputs the dataset 1 TB dataset with 5–7s latency. More fundamentally, to to a storage system). Like DryadLINQ, Spark computes illustrate the generality of RDDs, we have implemented RDDs lazily the first time they are used in an action, so the Pregel and HaLoop programming models on top of that it can pipeline transformations. Spark, including the placement optimizations they em- In addition, programmers can call a persist method to ploy, as relatively small libraries (200 lines of code each). indicate which RDDs they want to reuse in future oper- ations. Spark keeps persistent RDDs in memory by de- This paper begins with an overview of RDDs (§2) and fault, but it can spill them to disk if there is not enough Spark (§3). We then discuss the internal representation RAM. Users can also request other persistence strategies, of RDDs (§4), our implementation (§5), and experimen- such as storing the RDD only on disk or replicating it tal results (§6). Finally, we discuss how RDDs capture across machines, through flags to persist. Finally, users several existing cluster programming models (§7), sur- can set a persistence priority on each RDD to specify vey related work (§8), and conclude. which in-memory data should spill to disk first. 2 Resilient Distributed Datasets (RDDs) 2.2.1 Example: Console Log Mining This section provides an overview of RDDs. We first de- Suppose that a web service is experiencing errors and an fine RDDs (§2.1) and introduce their programming inter- operator wants to search terabytes of logs in the Hadoop face in Spark (§2.2). We then compare RDDs with finer- filesystem (HDFS) to find the cause. Using Spark, the op- grained shared memory abstractions (§2.3). Finally, we erator can load just the error messages from the logs into discuss limitations of the RDD model (§2.4). RAM across a set of nodes and query them interactively. 2.1 RDD Abstraction She would first type the following Scala code: 2 Although individual RDDs are immutable, it is possible to imple- Formally, an RDD is a read-only, partitioned collection ment mutable state by having multiple RDDs to represent multiple ver- of records. RDDs can only be created through determin- sions of a dataset. We made RDDs immutable to make it easier to de- istic operations on either (1) data in stable storage or (2) scribe lineage graphs, but it would have been equivalent to have our other RDDs. We call these operations transformations to abstraction be versioned datasets and track versions in lineage graphs. 

3 . lines Aspect RDDs Distr. Shared Mem. filter(_.startsWith(“ERROR”)) Reads Coarse- or fine-grained Fine-grained errors Writes Coarse-grained Fine-grained filter(_.contains(“HDFS”))) Consistency Trivial (immutable) Up to app / runtime HDFS errors Fault recovery Fine-grained and low- Requires checkpoints overhead using lineage and program rollback map(_.split(‘\t’)(3)) Straggler Possible using backup Difficult time fields mitigation tasks Figure 1: Lineage graph for the third query in our example. Work Automatic based on Up to app (runtimes placement data locality aim for transparency) Boxes represent RDDs and arrows represent transformations. Behavior if not Similar to existing data Poor performance enough RAM flow systems (swapping?) lines = spark.textFile("hdfs://...") errors = lines.filter(_.startsWith("ERROR")) Table 1: Comparison of RDDs with distributed shared memory. errors.persist() Line 1 defines an RDD backed by an HDFS file (as a 2.3 Advantages of the RDD Model collection of lines of text), while line 2 derives a filtered RDD from it. Line 3 then asks for errors to persist in To understand the benefits of RDDs as a distributed memory so that it can be shared across queries. Note that memory abstraction, we compare them against dis- the argument to filter is Scala syntax for a closure. tributed shared memory (DSM) in Table 1. In DSM sys- At this point, no work has been performed on the clus- tems, applications read and write to arbitrary locations in ter. However, the user can now use the RDD in actions, a global address space. Note that under this definition, we e.g., to count the number of messages: include not only traditional shared memory systems [24], but also other systems where applications make fine- errors.count() grained writes to shared state, including Piccolo [27], which provides a shared DHT, and distributed databases. The user can also perform further transformations on DSM is a very general abstraction, but this generality the RDD and use their results, as in the following lines: makes it harder to implement in an efficient and fault- tolerant manner on commodity clusters. // Count errors mentioning MySQL: The main difference between RDDs and DSM is that errors.filter(_.contains("MySQL")).count() RDDs can only be created (“written”) through coarse- // Return the time fields of errors mentioning grained transformations, while DSM allows reads and // HDFS as an array (assuming time is field writes to each memory location.3 This restricts RDDs // number 3 in a tab-separated format): to applications that perform bulk writes, but allows for errors.filter(_.contains("HDFS")) more efficient fault tolerance. In particular, RDDs do not .map(_.split(’\t’)(3)) need to incur the overhead of checkpointing, as they can .collect() be recovered using lineage.4 Furthermore, only the lost partitions of an RDD need to be recomputed upon fail- After the first action involving errors runs, Spark will ure, and they can be recomputed in parallel on different store the partitions of errors in memory, greatly speed- nodes, without having to roll back the whole program. ing up subsequent computations on it. Note that the base A second benefit of RDDs is that their immutable na- RDD, lines, is not loaded into RAM. This is desirable ture lets a system mitigate slow nodes (stragglers) by run- because the error messages might only be a small frac- ning backup copies of slow tasks as in MapReduce [10]. tion of the data (small enough to fit into memory). Backup tasks would be hard to implement with DSM, as Finally, to illustrate how our model achieves fault tol- the two copies of a task would access the same memory erance, we show the lineage graph for the RDDs in our locations and interfere with each other’s updates. third query in Figure 1. In this query, we started with Finally, RDDs provide two other benefits over DSM. errors, the result of a filter on lines, and applied a fur- First, in bulk operations on RDDs, a runtime can sched- ther filter and map before running a collect. The Spark scheduler will pipeline the latter two transformations and 3 Note that reads on RDDs can still be fine-grained. For example, an send a set of tasks to compute them to the nodes holding application can treat an RDD as a large read-only lookup table. 4 In some applications, it can still help to checkpoint RDDs with the cached partitions of errors. In addition, if a partition long lineage chains, as we discuss in Section 5.4. However, this can be of errors is lost, Spark rebuilds it by applying a filter on done in the background because RDDs are immutable, and there is no only the corresponding partition of lines. need to take a snapshot of the whole application as in DSM. 

4 . RAM tions like map by passing closures (function literals). Worker Scala represents each closure as a Java object, and Input Data these objects can be serialized and loaded on another RAM node to pass the closure across the network. Scala also Driver Worker saves any variables bound in the closure as fields in results the Java object. For example, one can write code like RAM Input Data tasks var x = 5; rdd.map(_ + x) to add 5 to each element Worker of an RDD.5 Input Data RDDs themselves are statically typed objects parametrized by an element type. For example, Figure 2: Spark runtime. The user’s driver program launches RDD[Int] is an RDD of integers. However, most of our multiple workers, which read data blocks from a distributed file examples omit types since Scala supports type inference. system and can persist computed RDD partitions in memory. Although our method of exposing RDDs in Scala is conceptually simple, we had to work around issues with ule tasks based on data locality to improve performance. Scala’s closure objects using reflection [33]. We also Second, RDDs degrade gracefully when there is not needed more work to make Spark usable from the Scala enough memory to store them, as long as they are only interpreter, as we shall discuss in Section 5.2. Nonethe- being used in scan-based operations. Partitions that do less, we did not have to modify the Scala compiler. not fit in RAM can be stored on disk and will provide 3.1 RDD Operations in Spark similar performance to current data-parallel systems. Table 2 lists the main RDD transformations and actions 2.4 Applications Not Suitable for RDDs available in Spark. We give the signature of each oper- As discussed in the Introduction, RDDs are best suited ation, showing type parameters in square brackets. Re- for batch applications that apply the same operation to call that transformations are lazy operations that define a all elements of a dataset. In these cases, RDDs can ef- new RDD, while actions launch a computation to return ficiently remember each transformation as one step in a a value to the program or write data to external storage. lineage graph and can recover lost partitions without hav- Note that some operations, such as join, are only avail- ing to log large amounts of data. RDDs would be less able on RDDs of key-value pairs. Also, our function suitable for applications that make asynchronous fine- names are chosen to match other APIs in Scala and other grained updates to shared state, such as a storage sys- functional languages; for example, map is a one-to-one tem for a web application or an incremental web crawler. mapping, while flatMap maps each input value to one or For these applications, it is more efficient to use systems more outputs (similar to the map in MapReduce). that perform traditional update logging and data check- In addition to these operators, users can ask for an pointing, such as databases, RAMCloud [25], Percolator RDD to persist. Furthermore, users can get an RDD’s [26] and Piccolo [27]. Our goal is to provide an efficient partition order, which is represented by a Partitioner programming model for batch analytics and leave these class, and partition another dataset according to it. Op- asynchronous applications to specialized systems. erations such as groupByKey, reduceByKey and sort au- tomatically result in a hash or range partitioned RDD. 3 Spark Programming Interface 3.2 Example Applications Spark provides the RDD abstraction through a language- integrated API similar to DryadLINQ [31] in Scala [2], We complement the data mining example in Section a statically typed functional programming language for 2.2.1 with two iterative applications: logistic regression the Java VM. We chose Scala due to its combination of and PageRank. The latter also showcases how control of conciseness (which is convenient for interactive use) and RDDs’ partitioning can improve performance. efficiency (due to static typing). However, nothing about 3.2.1 Logistic Regression the RDD abstraction requires a functional language. Many machine learning algorithms are iterative in nature To use Spark, developers write a driver program that because they run iterative optimization procedures, such connects to a cluster of workers, as shown in Figure 2. as gradient descent, to maximize a function. They can The driver defines one or more RDDs and invokes ac- thus run much faster by keeping their data in memory. tions on them. Spark code on the driver also tracks the As an example, the following program implements lo- RDDs’ lineage. The workers are long-lived processes gistic regression [14], a common classification algorithm that can store RDD partitions in RAM across operations. As we showed in the log mining example in Sec- 5 We save each closure at the time it is created, so that the map in tion 2.2.1, users provide arguments to RDD opera- this example will always add 5 even if x changes. 

5 . map( f : T ⇒ U) : RDD[T] ⇒ RDD[U] filter( f : T ⇒ Bool) : RDD[T] ⇒ RDD[T] flatMap( f : T ⇒ Seq[U]) : RDD[T] ⇒ RDD[U] sample(fraction : Float) : RDD[T] ⇒ RDD[T] (Deterministic sampling) groupByKey() : RDD[(K, V)] ⇒ RDD[(K, Seq[V])] reduceByKey( f : (V, V) ⇒ V) : RDD[(K, V)] ⇒ RDD[(K, V)] Transformations union() : (RDD[T], RDD[T]) ⇒ RDD[T] join() : (RDD[(K, V)], RDD[(K, W)]) ⇒ RDD[(K, (V, W))] cogroup() : (RDD[(K, V)], RDD[(K, W)]) ⇒ RDD[(K, (Seq[V], Seq[W]))] crossProduct() : (RDD[T], RDD[U]) ⇒ RDD[(T, U)] mapValues( f : V ⇒ W) : RDD[(K, V)] ⇒ RDD[(K, W)] (Preserves partitioning) sort(c : Comparator[K]) : RDD[(K, V)] ⇒ RDD[(K, V)] partitionBy(p : Partitioner[K]) : RDD[(K, V)] ⇒ RDD[(K, V)] count() : RDD[T] ⇒ Long collect() : RDD[T] ⇒ Seq[T] Actions reduce( f : (T, T) ⇒ T) : RDD[T] ⇒ T lookup(k : K) : RDD[(K, V)] ⇒ Seq[V] (On hash/range partitioned RDDs) save(path : String) : Outputs RDD to a storage system, e.g., HDFS Table 2: Transformations and actions available on RDDs in Spark. Seq[T] denotes a sequence of elements of type T. that searches for a hyperplane w that best separates two input file links ranks0 map sets of points (e.g., spam and non-spam emails). The al- join gorithm uses gradient descent: it starts w at a random contribs0 reduce + map value, and on each iteration, it sums a function of w over ranks1 the data to move w in a direction that improves it. contribs1 val points = spark.textFile(...) .map(parsePoint).persist() ranks2 var w = // random initial vector for (i <- 1 to ITERATIONS) { contribs2 val gradient = points.map{ p => p.x * (1/(1+exp(-p.y*(w dot p.x)))-1)*p.y . . . }.reduce((a,b) => a+b) Figure 3: Lineage graph for datasets in PageRank. w -= gradient } val links = spark.textFile(...).map(...).persist() We start by defining a persistent RDD called points var ranks = // RDD of (URL, rank) pairs as the result of a map transformation on a text file that for (i <- 1 to ITERATIONS) { parses each line of text into a Point object. We then re- // Build an RDD of (targetURL, float) pairs peatedly run map and reduce on points to compute the // with the contributions sent by each page gradient at each step by summing a function of the cur- val contribs = links.join(ranks).flatMap { rent w. Keeping points in memory across iterations can (url, (links, rank)) => links.map(dest => (dest, rank/links.size)) yield a 20× speedup, as we show in Section 6.1. } 3.2.2 PageRank // Sum contributions by URL and get new ranks A more complex pattern of data sharing occurs in ranks = contribs.reduceByKey((x,y) => x+y) .mapValues(sum => a/N + (1-a)*sum) PageRank [6]. The algorithm iteratively updates a rank } for each document by adding up contributions from doc- uments that link to it. On each iteration, each document This program leads to the RDD lineage graph in Fig- sends a contribution of nr to its neighbors, where r is its ure 3. On each iteration, we create a new ranks dataset rank and n is its number of neighbors. It then updates based on the contribs and ranks from the previous iter- its rank to α/N + (1 − α) ∑ ci , where the sum is over ation and the static links dataset.6 One interesting fea- the contributions it received and N is the total number of ture of this graph is that it grows longer with the number documents. We can write PageRank in Spark as follows: 6 Note that although RDDs are immutable, the variables ranks and // Load graph as an RDD of (URL, outlinks) pairs contribs in the program point to different RDDs on each iteration. 

6 .of iterations. Thus, in a job with many iterations, it may Operation Meaning be necessary to reliably replicate some of the versions partitions() Return a list of Partition objects of ranks to reduce fault recovery times [20]. The user preferredLocations(p) List nodes where partition p can be can call persist with a RELIABLE flag to do this. However, accessed faster due to data locality note that the links dataset does not need to be replicated, dependencies() Return a list of dependencies because partitions of it can be rebuilt efficiently by rerun- ning a map on blocks of the input file. This dataset will iterator(p, parentIters) Compute the elements of partition p given iterators for its parent partitions typically be much larger than ranks, because each docu- ment has many links but only one number as its rank, so partitioner() Return metadata specifying whether the RDD is hash/range partitioned recovering it using lineage saves time over systems that checkpoint a program’s entire in-memory state. Table 3: Interface used to represent RDDs in Spark. Finally, we can optimize communication in PageRank by controlling the partitioning of the RDDs. If we spec- of a map on this RDD has the same partitions, but applies ify a partitioning for links (e.g., hash-partition the link the map function to the parent’s data when computing its lists by URL across nodes), we can partition ranks in elements. We summarize this interface in Table 3. the same way and ensure that the join operation between The most interesting question in designing this inter- links and ranks requires no communication (as each face is how to represent dependencies between RDDs. URL’s rank will be on the same machine as its link list). We found it both sufficient and useful to classify depen- We can also write a custom Partitioner class to group dencies into two types: narrow dependencies, where each pages that link to each other together (e.g., partition the partition of the parent RDD is used by at most one parti- URLs by domain name). Both optimizations can be ex- tion of the child RDD, wide dependencies, where multi- pressed by calling partitionBy when we define links: ple child partitions may depend on it. For example, map links = spark.textFile(...).map(...) leads to a narrow dependency, while join leads to to wide .partitionBy(myPartFunc).persist() dependencies (unless the parents are hash-partitioned). Figure 4 shows other examples. After this initial call, the join operation between links This distinction is useful for two reasons. First, narrow and ranks will automatically aggregate the contributions dependencies allow for pipelined execution on one clus- for each URL to the machine that its link lists is on, cal- ter node, which can compute all the parent partitions. For culate its new rank there, and join it with its links. This example, one can apply a map followed by a filter on an type of consistent partitioning across iterations is one of element-by-element basis. In contrast, wide dependen- the main optimizations in specialized frameworks like cies require data from all parent partitions to be available Pregel. RDDs let the user express this goal directly. and to be shuffled across the nodes using a MapReduce- like operation. Second, recovery after a node failure is 4 Representing RDDs more efficient with a narrow dependency, as only the lost One of the challenges in providing RDDs as an abstrac- parent partitions need to be recomputed, and they can be tion is choosing a representation for them that can track recomputed in parallel on different nodes. In contrast, in lineage across a wide range of transformations. Ideally, a lineage graph with wide dependencies, a single failed a system implementing RDDs should provide as rich node might cause the loss of some partition from all the a set of transformation operators as possible (e.g., the ancestors of an RDD, requiring a complete re-execution. ones in Table 2), and let users compose them in arbitrary This common interface for RDDs made it possible to ways. We propose a simple graph-based representation implement most transformations in Spark in less than 20 for RDDs that facilitates these goals. We have used this lines of code. Indeed, even new Spark users have imple- representation in Spark to support a wide range of trans- mented new transformations (e.g., sampling and various formations without adding special logic to the scheduler types of joins) without knowing the details of the sched- for each one, which greatly simplified the system design. uler. We sketch some RDD implementations below. In a nutshell, we propose representing each RDD through a common interface that exposes five pieces of HDFS files: The input RDDs in our samples have been information: a set of partitions, which are atomic pieces files in HDFS. For these RDDs, partitions returns one of the dataset; a set of dependencies on parent RDDs; partition for each block of the file (with the block’s offset a function for computing the dataset based on its par- stored in each Partition object), preferredLocations gives ents; and metadata about its partitioning scheme and data the nodes the block is on, and iterator reads the block. placement. For example, an RDD representing an HDFS map: Calling map on any RDD returns a MappedRDD file has a partition for each block of the file and knows object. This object has the same partitions and preferred which machines each block is on. Meanwhile, the result locations as its parent, but applies the function passed to 

7 . Narrow Dependencies: Wide Dependencies: A: B: G: Stage 1 groupBy map, filter groupByKey C: D: F: map E: join join with inputs co-partitioned Stage 2 union Stage 3 join with inputs not union co-partitioned Figure 5: Example of how Spark computes job stages. Boxes Figure 4: Examples of narrow and wide dependencies. Each with solid outlines are RDDs. Partitions are shaded rectangles, box is an RDD, with partitions shown as shaded rectangles. in black if they are already in memory. To run an action on RDD G, we build build stages at wide dependencies and pipeline nar- map to the parent’s records in its iterator method. row transformations inside each stage. In this case, stage 1’s output RDD is already in RAM, so we run stage 2 and then 3. union: Calling union on two RDDs returns an RDD whose partitions are the union of those of the parents. Each child partition is computed through a narrow de- sistent RDDs are available in memory. Whenever a user pendency on the corresponding parent.7 runs an action (e.g., count or save) on an RDD, the sched- uler examines that RDD’s lineage graph to build a DAG sample: Sampling is similar to mapping, except that of stages to execute, as illustrated in Figure 5. Each stage the RDD stores a random number generator seed for each contains as many pipelined transformations with narrow partition to deterministically sample parent records. dependencies as possible. The boundaries of the stages join: Joining two RDDs may lead to either two nar- are the shuffle operations required for wide dependen- row dependencies (if they are both hash/range partitioned cies, or any already computed partitions that can short- with the same partitioner), two wide dependencies, or a circuit the computation of a parent RDD. The scheduler mix (if one parent has a partitioner and one does not). In then launches tasks to compute missing partitions from either case, the output RDD has a partitioner (either one each stage until it has computed the target RDD. inherited from the parents or a default hash partitioner). Our scheduler assigns tasks to machines based on data locality using delay scheduling [32]. If a task needs to 5 Implementation process a partition that is available in memory on a node, We have implemented Spark in about 14,000 lines of we send it to that node. Otherwise, if a task processes Scala. The system runs over the Mesos cluster man- a partition for which the containing RDD provides pre- ager [17], allowing it to share resources with Hadoop, ferred locations (e.g., an HDFS file), we send it to those. MPI and other applications. Each Spark program runs as For wide dependencies (i.e., shuffle dependencies), we a separate Mesos application, with its own driver (mas- currently materialize intermediate records on the nodes ter) and workers, and resource sharing between these ap- holding parent partitions to simplify fault recovery, much plications is handled by Mesos. like MapReduce materializes map outputs. Spark can read data from any Hadoop input source If a task fails, we re-run it on another node as long (e.g., HDFS or HBase) using Hadoop’s existing input as its stage’s parents are still available. If some stages plugin APIs, and runs on an unmodified version of Scala. have become unavailable (e.g., because an output from We now sketch several of the technically interesting the “map side” of a shuffle was lost), we resubmit tasks to parts of the system: our job scheduler (§5.1), our Spark compute the missing partitions in parallel. We do not yet interpreter allowing interactive use (§5.2), memory man- tolerate scheduler failures, though replicating the RDD agement (§5.3), and support for checkpointing (§5.4). lineage graph would be straightforward. 5.1 Job Scheduling Finally, although all computations in Spark currently Spark’s scheduler uses our representation of RDDs, de- run in response to actions called in the driver program, scribed in Section 4. we are also experimenting with letting tasks on the clus- Overall, our scheduler is similar to Dryad’s [19], but ter (e.g., maps) call the lookup operation, which provides it additionally takes into account which partitions of per- random access to elements of hash-partitioned RDDs by key. In this case, tasks would need to tell the scheduler to 7 Note that our union operation does not drop duplicate values. compute the required partition if it is missing. 

8 . Line1 String in-memory storage as serialized data, and on-disk stor- query: hello age. The first option provides the fastest performance, Line 1: var query = “hello” because the Java VM can access each RDD element Line2 natively. The second option lets users choose a more line1: memory-efficient representation than Java object graphs Line 2: rdd.filter(_.contains(query)) when space is limited, at the cost of lower performance.8 .count() Closure1 The third option is useful for RDDs that are too large to line1: keep in RAM but costly to recompute on each use. eval(s): { return s.contains(line1.query) } To manage the limited memory available, we use an LRU eviction policy at the level of RDDs. When a new a) Lines typed by user b) Resulting object graph RDD partition is computed but there is not enough space to store it, we evict a partition from the least recently ac- Figure 6: Example showing how the Spark interpreter translates cessed RDD, unless this is the same RDD as the one with two lines entered by the user into Java objects. the new partition. In that case, we keep the old partition in memory to prevent cycling partitions from the same 5.2 Interpreter Integration RDD in and out. This is important because most oper- Scala includes an interactive shell similar to those of ations will run tasks over an entire RDD, so it is quite Ruby and Python. Given the low latencies attained with likely that the partition already in memory will be needed in-memory data, we wanted to let users run Spark inter- in the future. We found this default policy to work well in actively from the interpreter to query big datasets. all our applications so far, but we also give users further control via a “persistence priority” for each RDD. The Scala interpreter normally operates by compiling Finally, each instance of Spark on a cluster currently a class for each line typed by the user, loading it into has its own separate memory space. In future work, we the JVM, and invoking a function on it. This class in- plan to investigate sharing RDDs across instances of cludes a singleton object that contains the variables or Spark through a unified memory manager. functions on that line and runs the line’s code in an ini- tialize method. For example, if the user types var x = 5 5.4 Support for Checkpointing followed by println(x), the interpreter defines a class Although lineage can always be used to recover RDDs called Line1 containing x and causes the second line to after a failure, such recovery may be time-consuming for compile to println(Line1.getInstance().x). RDDs with long lineage chains. Thus, it can be helpful We made two changes to the interpreter in Spark: to checkpoint some RDDs to stable storage. 1. Class shipping: To let the worker nodes fetch the In general, checkpointing is useful for RDDs with long bytecode for the classes created on each line, we lineage graphs containing wide dependencies, such as made the interpreter serve these classes over HTTP. the rank datasets in our PageRank example (§3.2.2). In 2. Modified code generation: Normally, the singleton these cases, a node failure in the cluster may result in object created for each line of code is accessed the loss of some slice of data from each parent RDD, re- through a static method on its corresponding class. quiring a full recomputation [20]. In contrast, for RDDs This means that when we serialize a closure refer- with narrow dependencies on data in stable storage, such encing a variable defined on a previous line, such as as the points in our logistic regression example (§3.2.1) Line1.x in the example above, Java will not trace and the link lists in PageRank, checkpointing may never through the object graph to ship the Line1 instance be worthwhile. If a node fails, lost partitions from these wrapping around x. Therefore, the worker nodes will RDDs can be recomputed in parallel on other nodes, at a not receive x. We modified the code generation logic fraction of the cost of replicating the whole RDD. to reference the instance of each line object directly. Spark currently provides an API for checkpointing (a REPLICATE flag to persist), but leaves the decision of Figure 6 shows how the interpreter translates a set of which data to checkpoint to the user. However, we are lines typed by the user to Java objects after our changes. also investigating how to perform automatic checkpoint- We found the Spark interpreter to be useful in process- ing. Because our scheduler knows the size of each dataset ing large traces obtained as part of our research and ex- as well as the time it took to first compute it, it should be ploring datasets stored in HDFS. We also plan to use to able to select an optimal set of RDDs to checkpoint to run higher-level query languages interactively, e.g., SQL. minimize system recovery time [30]. 5.3 Memory Management Finally, note that the read-only nature of RDDs makes Spark provides three options for storage of persistent 8 The cost depends on how much computation the application does RDDs: in-memory storage as deserialized Java objects, per byte of data, but can be up to 2× for lightweight processing. 

9 . 182! them simpler to checkpoint than general shared mem- 240! First Iteration! Iteration time (s)! Later Iterations! ory. Because consistency is not a concern, RDDs can be 200! 139! 115! written out in the background without requiring program 106! 160! 87! pauses or distributed snapshot schemes. 82! 120! 80! 76! 62! 46! 80! 33! 6 Evaluation 40! 3! We evaluated Spark and RDDs through a series of exper- 0! iments on Amazon EC2, as well as benchmarks of user Hadoop! HadoopBM! Spark! Hadoop! HadoopBM! Spark! Logistic Regression! K-Means! applications. Overall, our results show the following: • Spark outperforms Hadoop by up to 20× in itera- Figure 7: Duration of the first and later iterations in Hadoop, tive machine learning and graph applications. The HadoopBinMem and Spark for logistic regression and k-means speedup comes from avoiding I/O and deserialization using 100 GB of data on a 100-node cluster. costs by storing data in memory as Java objects. 274! 300! Hadoop! 300! Hadoop ! • Applications written by our users perform and scale HadoopBinMem! HadoopBinMem! Iteration time (s)! 197! 250! Spark! 250! Spark! Iteration time (s)! well. In particular, we used Spark to speed up an an- 184! 157! 200! 143! alytics report that was running on Hadoop by 40×. 200! 121! 116! 106! 111! 150! 150! • When nodes fail, Spark can recover quickly by re- 87! 80! 76! 62! 61! 100! 100! building only the lost RDD partitions. 33! 15! 50! 50! 6! 3! • Spark can be used to query a 1 TB dataset interac- 0! 0! tively with latencies of 5–7 seconds. 25! 50! 100! 25! 50! 100! Number of machines! Number of machines! We start by presenting benchmarks for iterative ma- chine learning applications (§6.1) and PageRank (§6.2) (a) Logistic Regression (b) K-Means against Hadoop. We then evaluate fault recovery in Spark Figure 8: Running times for iterations after the first in Hadoop, (§6.3) and behavior when a dataset does not fit in mem- HadoopBinMem, and Spark. The jobs all processed 100 GB. ory (§6.4). Finally, we discuss results for user applica- tions (§6.5) and interactive data mining (§6.6). Unless otherwise noted, our tests used m1.xlarge EC2 First Iterations All three systems read text input from nodes with 4 cores and 15 GB of RAM. We used HDFS HDFS in their first iterations. As shown in the light bars for storage, with 256 MB blocks. Before each test, we in Figure 7, Spark was moderately faster than Hadoop cleared OS buffer caches to measure IO costs accurately. across experiments. This difference was due to signal- ing overheads in Hadoop’s heartbeat protocol between 6.1 Iterative Machine Learning Applications its master and workers. HadoopBinMem was the slowest We implemented two iterative machine learning appli- because it ran an extra MapReduce job to convert the data cations, logistic regression and k-means, to compare the to binary, it and had to write this data across the network performance of the following systems: to a replicated in-memory HDFS instance. • Hadoop: The Hadoop 0.20.2 stable release. Subsequent Iterations Figure 7 also shows the aver- • HadoopBinMem: A Hadoop deployment that con- age running times for subsequent iterations, while Fig- verts the input data into a low-overhead binary format ure 8 shows how these scaled with cluster size. For lo- in the first iteration to eliminate text parsing in later gistic regression, Spark 25.3× and 20.7× faster than ones, and stores it in an in-memory HDFS instance. Hadoop and HadoopBinMem respectively on 100 ma- chines. For the more compute-intensive k-means appli- • Spark: Our implementation of RDDs. cation, Spark still achieved speedup of 1.9× to 3.2×. We ran both algorithms for 10 iterations on 100 GB Understanding the Speedup We were surprised to datasets using 25–100 machines. The key difference be- find that Spark outperformed even Hadoop with in- tween the two applications is the amount of computation memory storage of binary data (HadoopBinMem) by a they perform per byte of data. The iteration time of k- 20× margin. In HadoopBinMem, we had used Hadoop’s means is dominated by computation, while logistic re- standard binary format (SequenceFile) and a large block gression is less compute-intensive and thus more sensi- size of 256 MB, and we had forced HDFS’s data di- tive to time spent in deserialization and I/O. rectory to be on an in-memory file system. However, Since typical learning algorithms need tens of itera- Hadoop still ran slower due to several factors: tions to converge, we report times for the first iteration and subsequent iterations separately. We find that shar- 1. Minimum overhead of the Hadoop software stack, ing data via RDDs greatly speeds up future iterations. 2. Overhead of HDFS while serving data, and 

10 . 15.4! 171! 20! 13.1! Iteration time (s)! Text Input! 200! Iteration time (s)! 15! Binary Input! Hadoop! 150! 8.4! 80! 6.9! 72! 10! Basic Spark! 100! 2.9! 2.9! 28! 23! 14! 5! 50! Spark + Controlled Partitioning! 0! 0! In-mem HDFS! In-mem local file! Spark RDD! 30! 60! Number of machines! Figure 9: Iteration times for logistic regression using 256 MB data on a single machine for different sources of input. Figure 10: Performance of PageRank on Hadoop and Spark. 119! 140! No Failure! Iteratrion time (s)! 3. Deserialization cost to convert binary records to us- 120! Failure in the 6th Iteration! able in-memory Java objects. 81! 100! 59! 59! 58! 58! 57! 57! 57! 56! We investigated each of these factors in turn. To mea- 80! 60! sure (1), we ran no-op Hadoop jobs, and saw that these at 40! incurred least 25s of overhead to complete the minimal 20! requirements of job setup, starting tasks, and cleaning up. 0! 1! 2! 3! 4! 5! 6! 7! 8! 9! 10! Regarding (2), we found that HDFS performed multiple Iteration! memory copies and a checksum to serve each block. Finally, to measure (3), we ran microbenchmarks on Figure 11: Iteration times for k-means in presence of a failure. a single machine to run the logistic regression computa- One machine was killed at the start of the 6th iteration, resulting tion on 256 MB inputs in various formats. In particular, in partial reconstruction of an RDD using lineage. we compared the time to process text and binary inputs from both HDFS (where overheads in the HDFS stack 6.3 Fault Recovery will manifest) and an in-memory local file (where the kernel can very efficiently pass data to the program). We evaluated the cost of reconstructing RDD partitions We show the results of these tests in Figure 9. The dif- using lineage after a node failure in the k-means appli- ferences between in-memory HDFS and local file show cation. Figure 11 compares the running times for 10 it- that reading through HDFS introduced a 2-second over- erations of k-means on a 75-node cluster in normal op- head, even when data was in memory on the local ma- erating scenario, with one where a node fails at the start chine. The differences between the text and binary in- of the 6th iteration. Without any failure, each iteration put indicate the parsing overhead was 7 seconds. Finally, consisted of 400 tasks working on 100 GB of data. even when reading from an in-memory file, converting Until the end of the 5th iteration, the iteration times the pre-parsed binary data into Java objects took 3 sec- were about 58 seconds. In the 6th iteration, one of the onds, which is still almost as expensive as the logistic re- machines was killed, resulting in the loss of the tasks gression itself. By storing RDD elements directly as Java running on that machine and the RDD partitions stored objects in memory, Spark avoids all these overheads. there. Spark re-ran these tasks in parallel on other ma- chines, where they re-read corresponding input data and 6.2 PageRank reconstructed RDDs via lineage, which increased the it- We compared the performance of Spark with Hadoop eration time to 80s. Once the lost RDD partitions were for PageRank using a 54 GB Wikipedia dump. We ran reconstructed, the iteration time went back down to 58s. 10 iterations of the PageRank algorithm to process a Note that with a checkpoint-based fault recovery link graph of approximately 4 million articles. Figure 10 mechanism, recovery would likely require rerunning at demonstrates that in-memory storage alone provided least several iterations, depending on the frequency of Spark with a 2.4× speedup over Hadoop on 30 nodes. checkpoints. Furthermore, the system would need to In addition, controlling the partitioning of the RDDs to replicate the application’s 100 GB working set (the text make it consistent across iterations, as discussed in Sec- input data converted into binary) across the network, and tion 3.2.2, improved the speedup to 7.4×. The results would either consume twice the memory of Spark to also scaled nearly linearly to 60 nodes. replicate it in RAM, or would have to wait to write 100 We also evaluated a version of PageRank written us- GB to disk. In contrast, the lineage graphs for the RDDs ing our implementation of Pregel over Spark, which we in our examples were all less than 10 KB in size. describe in Section 7.1. The iteration times were similar 6.4 Behavior with Insufficient Memory to the ones in Figure 10, but longer by about 4 seconds because Pregel runs an extra operation on each iteration So far, we ensured that every machine in the cluster to let the vertices “vote” whether to finish the job. had enough memory to store all the RDDs across itera- 

11 . 100! 68.8! 70.6! Iteration time (s)! 58.1! 1521! 2000! 80! 80! Iteration time (s)! 40.7! Iteration time (s)! 1600! 29.7! 38.6! 60! 60! 820! 1200! 27.6! 11.5! 40! 40! 422! 800! 20! 400! 20! 0! 0%! 25%! 50%! 75%! 100%! 0! 0! 20! 40! 80! 20! 40! 80! Percent of dataset in memory! Number of machines! Number of machines! Figure 12: Performance of logistic regression using 100 GB (a) Traffic modeling (b) Spam classification data on 25 machines with varying amounts of data in memory. Figure 13: Per-iteration running time of two user applications implemented with Spark. Error bars show standard deviations. tions. A natural question is how Spark runs if there is not enough memory to store a job’s data. In this experiment, 10! Query response time (s)! Exact Match + View Count! 6.6! 7.0! we configured Spark not to use more than a certain per- Substring Match + View Count! 8! 5.5! Total View Count! 4.7! 4.5! centage of memory to store RDDs on each machine. We 6! present results for various amounts of storage space for 3.2! 2.8! logistic regression in Figure 12. We see that performance 4! 2.0! 1.7! degrades gracefully with less space. 2! 6.5 User Applications Built with Spark 0! 100 GB! 500 GB! 1 TB! In-Memory Analytics Conviva Inc, a video distribu- Data size (GB)! tion company, used Spark to accelerate a number of data Figure 14: Response times for interactive queries on Spark, analytics reports that previously ran over Hadoop. For scanning increasingly larger input datasets on 100 machines. example, one report ran as a series of Hive [1] queries that computed various statistics for a customer. These queries all worked on the same subset of the data (records Twitter Spam Classification The Monarch project at matching a customer-provided filter), but performed ag- Berkeley [29] used Spark to identify link spam in Twitter gregations (averages, percentiles, and COUNT DISTINCT) messages. They implemented a logistic regression classi- over different grouping fields, requiring separate MapRe- fier on top of Spark similar to the example in Section 6.1, duce jobs. By implementing the queries in Spark and but they used a distributed reduceByKey to sum the gradi- loading the subset of data shared across them once into ent vectors in parallel. In Figure 13(b) we show the scal- an RDD, the company was able to speed up the report by ing results for training a classifier over a 50 GB subset 40×. A report on 200 GB of compressed data that took of the data: 250,000 URLs and 107 features/dimensions 20 hours on a Hadoop cluster now runs in 30 minutes related to the network and content properties of the pages using only two Spark machines. Furthermore, the Spark at each URL. The scaling is not as close to linear due to program only required 96 GB of RAM, because it only a higher fixed communication cost per iteration. stored the rows and columns matching the customer’s fil- ter in an RDD, not the whole decompressed file. 6.6 Interactive Data Mining To demonstrate Spark’ ability to interactively query big Traffic Modeling Researchers in the Mobile Millen- datasets, we used it to analyze 1TB of Wikipedia page nium project at Berkeley [18] parallelized a learning al- view logs (2 years of data). For this experiment, we used gorithm for inferring road traffic congestion from spo- 100 m2.4xlarge EC2 instances with 8 cores and 68 GB radic automobile GPS measurements. The source data of RAM each. We ran queries to find total views of (1) were a 10,000 link road network for a metropolitan area, all pages, (2) pages with titles exactly matching a given as well as 600,000 samples of point-to-point trip times word, and (3) pages with titles partially matching a word. for GPS-equipped automobiles (travel times for each Each query scanned the entire input data. path may include multiple road links). Using a traffic model, the system can estimate the time it takes to travel Figure 14 shows the response times of the queries on across individual road links. The researchers trained this the full dataset and half and one-tenth of the data. Even model using an expectation maximization (EM) algo- at 1 TB of data, queries on Spark took 5–7 seconds. This rithm that repeats two map and reduceByKey steps itera- was more than an order of magnitude faster than work- tively. The application scales nearly linearly from 20 to ing with on-disk data; for example, querying the 1 TB 80 nodes with 4 cores each, as shown in Figure 13(a). file from disk took 170s. This illustrates that RDDs make Spark a powerful tool for interactive data mining. 

12 .7 Discussion RDD with the vertex states to perform the message ex- change. Equally importantly, RDDs allow us to keep ver- Although RDDs seem to offer a limited programming in- tex states in memory like Pregel does, to minimize com- terface due to their immutable nature and coarse-grained munication by controlling their partitioning, and to sup- transformations, we have found them suitable for a wide port partial recovery on failures. We have implemented class of applications. In particular, RDDs can express a Pregel as a 200-line library on top of Spark and refer the surprising number of cluster programming models that reader to [33] for more details. have so far been proposed as separate frameworks, al- lowing users to compose these models in one program Iterative MapReduce: Several recently proposed sys- (e.g., run a MapReduce operation to build a graph, then tems, including HaLoop [7] and Twister [11], provide an run Pregel on it) and share data between them. In this sec- iterative MapReduce model where the user gives the sys- tion, we discuss which programming models RDDs can tem a series of MapReduce jobs to loop. The systems express and why they are so widely applicable (§7.1). In keep data partitioned consistently across iterations, and addition, we discuss another benefit of the lineage infor- Twister can also keep it in memory. Both optimizations mation in RDDs that we are pursuing, which is to facili- are simple to express with RDDs, and we were able to tate debugging across these models (§7.2). implement HaLoop as a 200-line library using Spark. 7.1 Expressing Existing Programming Models Batched Stream Processing: Researchers have re- cently proposed several incremental processing systems RDDs can efficiently express a number of cluster pro- for applications that periodically update a result with gramming models that have so far been proposed inde- new data [21, 15, 4]. For example, an application updat- pendently. By “efficiently,” we mean that not only can ing statistics about ad clicks every 15 minutes should be RDDs be used to produce the same output as programs able to combine intermediate state from the previous 15- written in these models, but that RDDs can also capture minute window with data from new logs. These systems the optimizations that these frameworks perform, such as perform bulk operations similar to Dryad, but store appli- keeping specific data in memory, partitioning it to min- cation state in distributed filesystems. Placing the inter- imize communication, and recovering from failures effi- mediate state in RDDs would speed up their processing. ciently. The models expressible using RDDs include: Explaining the Expressivity of RDDs Why are RDDs MapReduce: This model can be expressed using the able to express these diverse programming models? The flatMap and groupByKey operations in Spark, or reduce- reason is that the restrictions on RDDs have little im- ByKey if there is a combiner. pact in many parallel applications. In particular, although DryadLINQ: The DryadLINQ system provides a RDDs can only be created through bulk transformations, wider range of operators than MapReduce over the more many parallel programs naturally apply the same opera- general Dryad runtime, but these are all bulk operators tion to many records, making them easy to express. Sim- that correspond directly to RDD transformations avail- ilarly, the immutability of RDDs is not an obstacle be- able in Spark (map, groupByKey, join, etc). cause one can create multiple RDDs to represent versions SQL: Like DryadLINQ expressions, SQL queries per- of the same dataset. Indeed, many of today’s MapReduce form data-parallel operations on sets of records. applications run over filesystems that do not allow up- dates to files, such as HDFS. Pregel: Google’s Pregel [22] is a specialized model for One final question is why previous frameworks have iterative graph applications that at first looks quite differ- not offered the same level of generality. We believe that ent from the set-oriented programming models in other this is because these systems explored specific problems systems. In Pregel, a program runs as a series of coordi- that MapReduce and Dryad do not handle well, such as nated “supersteps.” On each superstep, each vertex in the iteration, without observing that the common cause of graph runs a user function that can update state associ- these problems was a lack of data sharing abstractions. ated with the vertex, change the graph topology, and send messages to other vertices for use in the next superstep. 7.2 Leveraging RDDs for Debugging This model can express many graph algorithms, includ- While we initially designed RDDs to be deterministically ing shortest paths, bipartite matching, and PageRank. recomputable for fault tolerance, this property also facil- The key observation that lets us implement this model itates debugging. In particular, by logging the lineage of with RDDs is that Pregel applies the same user function RDDs created during a job, one can (1) reconstruct these to all the vertices on each iteration. Thus, we can store the RDDs later and let the user query them interactively and vertex states for each iteration in an RDD and perform (2) re-run any task from the job in a single-process de- a bulk transformation (flatMap) to apply this function bugger, by recomputing the RDD partitions it depends and generate an RDD of messages. We can then join this on. Unlike traditional replay debuggers for general dis- 

13 .tributed systems [13], which must capture or infer the and key-value stores like RAMCloud [25] offer a simi- order of events across multiple nodes, this approach adds lar model. RDDs differ from these systems in two ways. virtually zero recording overhead because only the RDD First, RDDs provide a higher-level programming inter- lineage graph needs to be logged.9 We are currently de- face based on operators such as map, sort and join, veloping a Spark debugger based on these ideas [33]. whereas the interface in Piccolo and DSM is just reads and updates to table cells. Second, Piccolo and DSM sys- 8 Related Work tems implement recovery through checkpoints and roll- Cluster Programming Models: Related work in clus- back, which is more expensive than the lineage-based ter programming models falls into several classes. First, strategy of RDDs in many applications. Finally, as dis- data flow models such as MapReduce [10], Dryad [19] cussed in Section 2.3, RDDs also provide other advan- and Ciel [23] support a rich set of operators for pro- tages over DSM, such as straggler mitigation. cessing data but share it through stable storage systems. Caching Systems: Nectar [12] can reuse intermediate RDDs represent a more efficient data sharing abstraction results across DryadLINQ jobs by identifying common than stable storage because they avoid the cost of data subexpressions with program analysis [16]. This capabil- replication, I/O and serialization.10 ity would be compelling to add to an RDD-based system. Second, several high-level programming interfaces However, Nectar does not provide in-memory caching (it for data flow systems, including DryadLINQ [31] and places the data in a distributed file system), nor does it FlumeJava [8], provide language-integrated APIs where let users explicitly control which datasets to persist and the user manipulates “parallel collections” through op- how to partition them. Ciel [23] and FlumeJava [8] can erators like map and join. However, in these systems, likewise cache task results but do not provide in-memory the parallel collections represent either files on disk or caching or explicit control over which data is cached. ephemeral datasets used to express a query plan. Al- though the systems will pipeline data across operators Ananthanarayanan et al. have proposed adding an in- in the same query (e.g., a map followed by another memory cache to distributed file systems to exploit the map), they cannot share data efficiently across queries. temporal and spatial locality of data access [3]. While We based Spark’s API on the parallel collection model this solution provides faster access to data that is already due to its convenience, and do not claim novelty for the in the file system, it is not as efficient a means of shar- language-integrated interface, but by providing RDDs as ing intermediate results within an application as RDDs, the storage abstraction behind this interface, we allow it because it would still require applications to write these to support a far broader class of applications. results to the file system between stages. A third class of systems provide high-level interfaces Lineage: Capturing lineage or provenance information for specific classes of applications requiring data sharing. for data has long been a research topic in scientific com- For example, Pregel [22] supports iterative graph appli- puting and databases, for applications such as explaining cations, while Twister [11] and HaLoop [7] are iterative results, allowing them to be reproduced by others, and MapReduce runtimes. However, these frameworks per- recomputing data if a bug is found in a workflow or if form data sharing implicitly for the pattern of computa- a dataset is lost. We refer the reader to [5] and [9] for tion they support, and do not provide a general abstrac- surveys of this work. RDDs provide a parallel program- tion that the user can employ to share data of her choice ming model where fine-grained lineage is inexpensive to among operations of her choice. For example, a user can- capture, so that it can be used for failure recovery. not use Pregel or Twister to load a dataset into memory Our lineage-based recovery mechanism is also similar and then decide what query to run on it. RDDs provide to the recovery mechanism used within a computation a distributed storage abstraction explicitly and can thus (job) in MapReduce and Dryad, which track dependen- support applications that these specialized systems do cies among a DAG of tasks. However, in these systems, not capture, such as interactive data mining. the lineage information is lost after a job ends, requiring Finally, some systems expose shared mutable state the use of a replicated storage system to share data across to allow the user to perform in-memory computation. computations. In contrast, RDDs apply lineage to persist For example, Piccolo [27] lets users run parallel func- in-memory data efficiently across computations, without tions that read and update cells in a distributed hash the cost of replication and disk I/O. table. Distributed shared memory (DSM) systems [24] Relational Databases: RDDs are conceptually similar 9 Unlike these systems, an RDD-based debugger will not replay non- to views in a database, and persistent RDDs resemble deterministic behavior in the user’s functions (e.g., a nondeterministic materialized views [28]. However, like DSM systems, map), but it can at least report it by checksumming data. 10 Note that running MapReduce/Dryad over an in-memory data store databases typically allow fine-grained read-write access like RAMCloud [25] would still require data replication and serializa- to all records, requiring logging of operations and data tion, which can be costly for some applications, as shown in §6.1. for fault tolerance and additional overhead to maintain 

14 .consistency. These overheads are not required with the [11] J. Ekanayake, H. Li, B. Zhang, T. Gunarathne, S.-H. Bae, J. Qiu, coarse-grained transformation model of RDDs. and G. Fox. Twister: a runtime for iterative mapreduce. In HPDC ’10, 2010. 9 Conclusion [12] P. K. Gunda, L. Ravindranath, C. A. Thekkath, Y. Yu, and L. Zhuang. Nectar: automatic management of data and We have presented resilient distributed datasets (RDDs), computation in datacenters. In OSDI ’10, 2010. an efficient, general-purpose and fault-tolerant abstrac- [13] Z. Guo, X. Wang, J. Tang, X. Liu, Z. Xu, M. Wu, M. F. Kaashoek, and Z. Zhang. R2: an application-level kernel for tion for sharing data in cluster applications. RDDs can record and replay. OSDI’08, 2008. express a wide range of parallel applications, including [14] T. Hastie, R. Tibshirani, and J. Friedman. The Elements of many specialized programming models that have been Statistical Learning: Data Mining, Inference, and Prediction. proposed for iterative computation, and new applications Springer Publishing Company, New York, NY, 2009. [15] B. He, M. Yang, Z. Guo, R. Chen, B. Su, W. Lin, and L. Zhou. that these models do not capture. Unlike existing storage Comet: batched stream processing for data intensive distributed abstractions for clusters, which require data replication computing. In SoCC ’10. for fault tolerance, RDDs offer an API based on coarse- [16] A. Heydon, R. Levin, and Y. Yu. Caching function calls using precise dependencies. In ACM SIGPLAN Notices, pages grained transformations that lets them recover data ef- 311–320, 2000. ficiently using lineage. We have implemented RDDs in [17] B. Hindman, A. Konwinski, M. Zaharia, A. Ghodsi, A. D. a system called Spark that outperforms Hadoop by up Joseph, R. H. Katz, S. Shenker, and I. Stoica. Mesos: A platform to 20× in iterative applications and can be used interac- for fine-grained resource sharing in the data center. In NSDI ’11. [18] T. Hunter, T. Moldovan, M. Zaharia, S. Merzgui, J. Ma, M. J. tively to query hundreds of gigabytes of data. Franklin, P. Abbeel, and A. M. Bayen. Scaling the Mobile We have open sourced Spark at spark-project.org as Millennium system in the cloud. In SOCC ’11, 2011. a vehicle for scalable data analysis and systems research. [19] M. Isard, M. Budiu, Y. Yu, A. Birrell, and D. Fetterly. Dryad: distributed data-parallel programs from sequential building Acknowledgements blocks. In EuroSys ’07, 2007. [20] S. Y. Ko, I. Hoque, B. Cho, and I. Gupta. On availability of We thank the first Spark users, including Tim Hunter, intermediate data in cloud computations. In HotOS ’09, 2009. Lester Mackey, Dilip Joseph, and Jibin Zhan, for trying [21] D. Logothetis, C. Olston, B. Reed, K. C. Webb, and K. Yocum. out our system in their real applications, providing many Stateful bulk processing for incremental analytics. SoCC ’10. [22] G. Malewicz, M. H. Austern, A. J. Bik, J. C. Dehnert, I. Horn, good suggestions, and identifying a few research chal- N. Leiser, and G. Czajkowski. Pregel: a system for large-scale lenges along the way. We also thank our shepherd, Ed graph processing. In SIGMOD, 2010. Nightingale, and our reviewers for their feedback. This [23] D. G. Murray, M. Schwarzkopf, C. Smowton, S. Smith, A. Madhavapeddy, and S. Hand. Ciel: a universal execution research was supported in part by Berkeley AMP Lab engine for distributed data-flow computing. In NSDI, 2011. sponsors Google, SAP, Amazon Web Services, Cloud- [24] B. Nitzberg and V. Lo. Distributed shared memory: a survey of era, Huawei, IBM, Intel, Microsoft, NEC, NetApp and issues and algorithms. Computer, 24(8):52 –60, Aug 1991. VMWare, by DARPA (contract #FA8650-11-C-7136), [25] J. Ousterhout, P. Agrawal, D. Erickson, C. Kozyrakis, J. Leverich, D. Mazi`eres, S. Mitra, A. Narayanan, G. Parulkar, by a Google PhD Fellowship, and by the Natural Sci- M. Rosenblum, S. M. Rumble, E. Stratmann, and R. Stutsman. ences and Engineering Research Council of Canada. The case for RAMClouds: scalable high-performance storage entirely in DRAM. SIGOPS Op. Sys. Rev., 43:92–105, Jan 2010. References [26] D. Peng and F. Dabek. Large-scale incremental processing using [1] Apache Hive. http://hadoop.apache.org/hive. distributed transactions and notifications. In OSDI 2010. [2] Scala. http://www.scala-lang.org. [27] R. Power and J. Li. Piccolo: Building fast, distributed programs [3] G. Ananthanarayanan, A. Ghodsi, S. Shenker, and I. Stoica. with partitioned tables. In Proc. OSDI 2010, 2010. Disk-locality in datacenter computing considered irrelevant. In [28] R. Ramakrishnan and J. Gehrke. Database Management HotOS ’11, 2011. Systems. McGraw-Hill, Inc., 3 edition, 2003. [4] P. Bhatotia, A. Wieder, R. Rodrigues, U. A. Acar, and [29] K. Thomas, C. Grier, J. Ma, V. Paxson, and D. Song. Design and R. Pasquin. Incoop: MapReduce for incremental computations. evaluation of a real-time URL spam filtering service. In IEEE In ACM SOCC ’11, 2011. Symposium on Security and Privacy, 2011. [5] R. Bose and J. Frew. Lineage retrieval for scientific data [30] J. W. Young. A first order approximation to the optimum processing: a survey. ACM Computing Surveys, 37:1–28, 2005. checkpoint interval. Commun. ACM, 17:530–531, Sept 1974. [31] Y. Yu, M. Isard, D. Fetterly, M. Budiu, U.´ Erlingsson, P. K. [6] S. Brin and L. Page. The anatomy of a large-scale hypertextual web search engine. In WWW, 1998. Gunda, and J. Currey. DryadLINQ: A system for [7] Y. Bu, B. Howe, M. Balazinska, and M. D. Ernst. HaLoop: general-purpose distributed data-parallel computing using a efficient iterative data processing on large clusters. Proc. VLDB high-level language. In OSDI ’08, 2008. Endow., 3:285–296, September 2010. [32] M. Zaharia, D. Borthakur, J. Sen Sarma, K. Elmeleegy, [8] C. Chambers, A. Raniwala, F. Perry, S. Adams, R. R. Henry, S. Shenker, and I. Stoica. Delay scheduling: A simple technique R. Bradshaw, and N. Weizenbaum. FlumeJava: easy, efficient for achieving locality and fairness in cluster scheduling. In data-parallel pipelines. In PLDI ’10. ACM, 2010. EuroSys ’10, 2010. [9] J. Cheney, L. Chiticariu, and W.-C. Tan. Provenance in [33] M. Zaharia, M. Chowdhury, T. Das, A. Dave, J. Ma, databases: Why, how, and where. Foundations and Trends in M. McCauley, M. Franklin, S. Shenker, and I. Stoica. Resilient Databases, 1(4):379–474, 2009. distributed datasets: A fault-tolerant abstraction for in-memory [10] J. Dean and S. Ghemawat. MapReduce: Simplified data cluster computing. Technical Report UCB/EECS-2011-82, processing on large clusters. In OSDI, 2004. EECS Department, UC Berkeley, 2011.