本文提出了分布式内存抽象的概念——弹性分布式数据集（RDD，Resilient Distributed Datasets）。它同意开发者在大型集群上运行基于内存的计算。RDD适用于两种应用，而现有的数据流系统对这两种应用的处理并不高效：一是迭代式算法，这在图应用和机器学习领域非经常见。二是交互式数据挖掘工具。这两种情况下。将数据保存在内存中可以极大地提高性能。为了有效地实现容错，RDD提供了一种高度受限的共享内存，即RDD在共享状态的时候是基于粗粒度的转换而不是细粒度的更新（换句话说就是，RDD仅仅能通过其它RDD上的批量操作来创建）。

虽然如此，RDD仍然足以用于非常多类型的计算。包含专用的迭代编程模型（如Pregel）等。我们已经将RDD应用到Spark系统之中，并对改性进行了相关的评估。

    像MapReduce和Dryad这种集群计算框架已经被广泛地应用到了大规模的数据分析中。这些系统提供了高级的运算符（算子），使得用户可以轻松地写出并行计算的程序而不用考虑任务分配问题以及错误容忍的细节。

    虽然眼下的框架都提供使用集群计算资源的高级抽象，可是使用分布式内存却没有相应的类似的高级抽象。

假设不能非常好地使用内存。那么，当一些应用的数据结果须要多次复用的情况出现的时候，运行效率就会受到影响。这些应用一般包含：

（1）迭代式机器学习和图应用中经常使用的迭代算法（每一步对数据运行相似的函数），比方pagerank、K-means聚合、逻辑回归等算法。

（2）交互式数据挖掘工具（用户重复查询一个数据子集）。

   当前的框架採取的方法是将须要复用的数据存储到存储系统中（HDFS等），然后在每次查询时又一次载入，那么，在数据复制、磁盘I/O、序列化的过程中会产生较大的开销，而这些开销会支配整个任务运行的时间。

    针对这个问题。我们提出了RDD，它能够为大部分的应用提供高效的数据复用。RDD是一个具有容错机制，并行数据架构，能够让用户将中间结果保存在内存中。并通过控制RDD的数据分区进而优化数据部署，还提供丰富的算子来操纵RDDs。

    就错误容忍而言，如今的集群系统採取的方式通常是提供细粒度的更新操作：数据备份或者是通过log处理进行数据恢复。对于数据密集型的任务来说。由于要进行大量的数据拷贝，network的带宽压力非常大（network的带宽小于RAM（内存cache）的带宽），且文件系统负载也会加大。

    相比而言，RDD提供了一种粗粒度的转换方式来解决容物容忍的问题。RDD通过Lineage（包括了怎样从其它RDD衍生所必需的相关信息），从而不须要检查点操作就能够重构丢失的数据分区。

    我们通过微基准和用户应用程序来评估RDD。实验表明。在处理迭代式应用上Spark比Hadoop快高达20多倍，计算数据分析类报表的性能提高了40多倍，同一时候可以在5-7秒的延时内交互式扫描1TB数据集。

此外。我们还在Spark之上实现了Pregel和HaLoop编程模型（包含其位置优化策略），以库的形式实现（分别使用了100和200行Scala代码）。

• RDD Abstraction抽像

    最后要说明的是，用户能够控制的RDDs的其它两个方面，持久化和划分：

    （1）用户能够指定须要复用的RDDs并选择一个基于内存的存储策略。

用户能够请求将RDD缓存，这样执行时将已经计算好的RDD分区存储起来，以加速后期的重用。相关策略：缓存的RDD默认存储在内存中，但假设内存不够，能够写到磁盘上；用户也能够使用其它的存储策略。比方只将RDD存储在磁盘等等；用户还能够在每一个RDD上面指定优先级来确定最先spill到磁盘的RDD。

    （2）用户能够就RDD每条记录里面的key进行划分。

这有助于部署的优化。比如，我们能够将两个数据集用相同的方式进行 hash-partitioned以便于之后的join操作（事先经过key进行了划分）。

• Spark Programming Interface

    在Spark中，RDD被表示为对象，通过这些对象上的方法（或函数）的调用在其上进行转换操作。

    程序猿一開始通过在存储层上的数据“转换”（map、filter等）定义一个或者多个RDDs。即transformations算子操作过程。之后。程序猿就能够在“动作”（actions）中使用RDD了。

动作是向应用程序返回值。或向存储系统导出数据的那些操作，比如，count（返回RDD中的元素个数）。collect（返回元素本身），save（将RDD输出到存储系统）。

在Spark中，仅仅有在RDD上第一次运行actions 操作时，才会真正開始计算RDD（即延迟计算）。

    【补充】每一个运算（如flatMap、map）事实上返回的都是一个RDD对象。每一个RDD对象都有一个Parent，通过这个Parent，实际上我们把一个个RDD对象串联了起来！能够觉得最后形成了一个RDD对象的队列。直到最后须要计算时（调用了action算子，后调用runjob函数）才開始逐一调用各个RDD对象的compute方法。完毕实际的运算。

    

    Example: Console Log Mining（控制台的日志挖掘）

    本部分我们通过一个详细演示样例来阐述RDD。

假定有一个大型站点出错，操作员想要检查Hadoop文件系统（HDFS）中的日志文件（TB级大小）来找出原因。通过使用Spark。操作员仅仅需将日志中的错误信息装载到一组节点的内存中，然后运行交互式查询就可以。

    首先，须要在Spark解释器中输入例如以下Scala命令：

    <strong>lines </strong>= spark.textFile("hdfs://...")    errors = lines.filter(_.startsWith("ERROR"))

    errors.cache()

     第1行从HDFS文件定义了一个RDD（即一个文本行集合），第2行获得一个过滤后的RDD，第3行请求将errors缓存起来。注意在Scala语法中filter的參数是一个闭包。

    errors.count()

    

    用户还能够在RDD上运行很多其它的转换操作。并使用转换结果，如

<span style="font-size:18px;">    // Count errors mentioning MySQL

    errors.filter(_.contains("MySQL")).count()

    // Return the time fields of errors mentioning

    // HDFS as an array (assuming time is field

    // number 3 in a tab-separated format):

    errors.filter(_.contains("HDFS"))

	.map(_.split('\t')(3))

	.collect()</span>

    使用errors的第一个action执行以后。Spark会把errors的分区缓存在内存中（action之后才会将数据载入到内存）。极大地加快了兴许计算速度。注意，最初的RDD
lines由于错误信息可能仅仅占原数据集的非常小一部分（小到足以放入内存）。

    最后。为了说明模型的容错性，下图给出了上述查询过程的的Lineage图。在RDD lines上运行filter操作，得到errors。然后再filter、map后得到新的RDD。在这个RDD上运行collect操作。

Spark调度器以流水线的方式运行后两个转换，向拥有errors分区缓存的节点发送一组任务。此外。假设某个errors分区丢失，Spark仅仅在对应的lines分区上运行filter操作（“血统”信息记录操作过程）来重建该errors分区。

    

• Advantages of the RDD Model（分布式共享内存）

    

    在DSM系统中。应用能够向全局地址空间的任何位置进行读写操作。（注意这里的DSM，不仅指传统的共享内存系统，还包含那些通过分布式哈希表或分布式文件系统进行细粒度的数据共享的系统，比方Piccolo）DSM是一种通用的抽象，但这样的通用性同一时候也使得在集群上实现有效的容错性更加困难。

    RDD与DSM主要差别在于，RDD仅仅能通过粗粒度的批量转换来创建（即“写”）。然而，DSM能够对随意内存位置读写。也就是说，RDD限制应用仅仅能运行批量写操作，这样有利于实现有效的容错。特别地，RDD没有设置检查点的开销。由于能够使用Lineage来恢复RDD。

并且。失效时仅仅须要又一次计算丢失的那些RDD分区。能够在不同节点上并行运行，而不须要回滚整个程序。

    【注意】通过内存中的任务备份，RDD还能够处理落后任务（即执行非常慢的节点），这点与MapReduce类似。而DSM则难以实现备份任务。由于任务及其副本都须要读写同一个内存位置。

    与DSM相比，RDD模型有两个优点。

第一，对于RDD中的批量操作，执行时将依据数据存放的位置来调度任务，从而提高性能。第二。对于基于扫描的操作。假设内存不足以缓存整个RDD。就进行部分缓存。

把内存放不下的分区存储到磁盘上。此时性能与现有的数据并行系统差点儿相同。

    最后看一下读操作的粒度。

RDD上的非常多动作（如count和collect）都是批量读操作，即扫描整个数据集，能够将任务分配到距离数据近期的节点上。同一时候。RDD也支持细粒度操作，即在哈希或范围分区的RDD上运行keyword查找。

• Applications Not Suitable for RDDs

Spark Programming Interface

    Spark用Scala语言实现了RDD的API。Scala是一种基于JVM的静态类型、函数式、面向对象的语言。

我们选择Scala是由于它简洁（特别适合交互式使用）、有效（由于是静态类型）。

可是，RDD抽象并不局限于函数式语言。

    为了使用Spark，开发人员须要编写一个driver程序，连接到集群以执行Worker，例如以下图所看到的。Driver定义了一个或多个RDD，并调用RDD上的动作。

Worker是长时间执行的进程。将RDD分区以Java对象的形式缓存在内存中。

    

    再看看上面对log文件信息进行操作的样例，用户运行RDD操作时会提供參数，比方map会传递一个闭包（closure。函数式编程中的概念）。

Scala将闭包表示为Java对象。这些对象在传递的时候被序列化，通过网络传输到其它节点上进行装载。Scala将闭包内的变量保存为Java对象的字段。比如。var
x = 5; rdd.map(_ + x) 这段代码将RDD中的每一个元素加5。

    RDD本身是静态类型对象。由參数指定其元素类型。比如，RDD[int]是一个整型RDD。

只是。我们举的样例差点儿都省略了这个类型參数。由于Scala支持类型判断。

    尽管在概念上使用Scala实现RDD非常easy，但还是要处理一些Scala闭包对象的反射问题，我们也须要解决如何通过Scala解释器来使用Spark还须要很多其它工作。

无论如何，我们都不须要改动Scala编译器。

• RDD Operations in Spark

    

    【注意】有些操作仅仅对键值对可用，比方join。另外。函数名与Scala及其它函数式语言中的API匹配。比如map是一对一的映射，而flatMap是将每一个输入映射为一个或多个输出（与MapReduce中的map类似）。

    除了上述这些操作以外。用户还能够请求将RDD缓存起来。并且。用户还能够通过Partitioner类获取RDD的分区顺序，然后将还有一个RDD依照相同的方式分区。还有些操作会自己主动产生一个哈希或范围分区的RDD。像groupByKey，reduceByKey和sort等。

• Example Applications

    Logistic Regression（逻辑规约）

    非常多机器学习算法都具有迭代特性，执行迭代优化方法来优化某个目标函数，比如梯度下降方法。

假设这些算法的工作集可以放入内存，将极大地加速程序执行。

    比如以下的程序是逻辑回归（一种常见的分类算法）的实现，该程序寻找一个最佳切割两组点（即垃圾邮件和非垃圾邮件）的超平面w。

算法採用梯度下降的方法：開始时w为随机值，在每一次迭代的过程中，对w的函数求和。然后朝着优化的方向移动w。

<span style="font-size:18px;">val points = spark.textFile(...)

         .map(parsePoint).persist()

    var w =   // random initial vector

    for (i <- 1 to ITERATIONS) {

         val gradient = points.map{ p =>

              p.x * (1/(1+exp(-p.y*(w dot p.x)))-1)*p.y

         }.reduce((a,b) => a+b)

         w -= gradient

    }</span>

    上述程序首先定义一个名为points的缓存RDD。这是在文本文件上运行map转换之后得到的，即将每一个文本行解析为一个Point对象。然后在points上重复运行map和reduce操作。每次迭代时通过对当前w的函数进行求和来计算梯度。

之后的小节我们将看到这样的在内存中缓存points的方式，比每次迭代都从磁盘文件装载数据并进行解析要快20X。

    已经在Spark中实现的迭代式机器学习算法还有：kmeans（像逻辑回归一样每次迭代时运行一对map和reduce操作）。期望最大化算法（EM，两个不同的map/reduce步骤交替运行）。交替最小二乘矩阵分解和协同过滤算法。

• PageRank

<span style="font-size:18px;">// Load graph as an RDD of (URL, outlinks) pairs

    val links = spark.textFile(...).map(...).persist()

    var ranks =    // RDD of (URL, rank) pairs

    for (i <- 1 to ITERATIONS) {

	// Build an RDD of (targetURL, float) pairs

	// with the contributions sent by each page

	val contribs = links.join(ranks).flatMap {

    	    (url, (links, rank)) =>

	        links.map(dest => (dest, rank/links.size))

	}

	// Sum contributions by URL and get new ranks

	ranks = contribs.reduceByKey((x,y) => x+y)

	                    .mapValues(sum => a/N + (1-a)*sum)

    }</span>

    该程序相应的“血统”图例如以下：

    

    在上图中，每一次的迭代。我们都会在之前的迭代过程以及静态的links数据集的基础上创建新的ranks数据集。这个图的一个有趣的功能是它的规模会随着迭代的次数增长。因此。在一个有非常多次迭代的job中，对ranks进行复制备份进而降低错误恢复的次数也是非常必要的。然而。须要注意的是links数据集不须要进行复制备份。由于links的恢复能够通过“血统”来高效地运行（对文件进行map操作）。links的数据集会比ranks大非常多，由于非常多文件都有非常多link可是。一个文件仅有一个rank。

    最后一点，我们能够通过控制RDD的划分来优化PageRank处理过程中的交互。假设我们指明了links的划分（hash-partition）。我们能够用同样的方法划分ranks的数据集而且能够确定links和ranks之间的join操作不须要交互（由于每个URL的link和它的link列表都在同一个机器（由于hash-partition））。我们也能够写一个自己定义的Partitioner类来聚集相互链接的pages（比方，通过域名划分）。上述的每种优化过程都能够在定义links的时候通过调用partitionBy来进行：

<span style="font-size:18px;">    links = spark.textFile(...).map(...)

                      .partitionBy(myPartFunc).persist()</span>

    通过上述的调用，links和ranks之间的join操作将会自己主动地获取到来自其它page对当前页面的贡献并又一次计算当前页面的rank值。

Representing RDDs

    提供RDD作为抽象的一个挑战是针对通过大范围的转换进行跟踪的“血统”的表示方式的选择。

理想的话，一个运行RDDs相关操作的系统应该尽可能多的提供转换算子，而且同意用户随意组合这些算子。关于上述的挑战，一个简单的基于图的RDDs表示法有助于目标的完毕。

在Spark中已经開始使用。

    概括来说，每一个RDD须要提供包括四种类型信息的接口：

    （1）一组RDD分区（partition。即数据集的原子组成部分）；

    （2）对父RDD的一组依赖，这些依赖描写叙述了RDD的Lineage；

    （3）一个函数，即在父RDD上运行何种计算；

    （4）元数据。描写叙述分区模式和数据存放的位置。比如。一个表示HDFS文件的RDD包括：各个数据块的一个分区。并知道各个数据块放在哪些节点上。并且，这个RDD上的map操作结果也具有相同的分区，map函数是在父数据上运行的。下表总结了RDD的内部接口。

    设计接口的一个关键问题就是，怎样表示RDD之间的依赖。我们发现RDD之间的依赖关系能够分为两类，即：（1）窄依赖（narrow dependencies）：子RDD的每一个分区至多依赖一个父分区（即与数据规模无关）。（2）宽依赖（wide
dependencies）：子RDD的多个分区能够依赖一个父RDD分区。比如。map产生窄依赖，而join则是宽依赖（除非父RDD被哈希分区）。还有一个样例见下图：

    

    区分这两种依赖非常实用。

首先，窄依赖同意在一个集群节点上以流水线的方式（pipeline）计算全部父分区。比如。逐个元素地运行map、然后filter操作；而宽依赖则须要获取全部父分区数据，然后在节点之间进行Shuffle，这与MapReduce类似。第二，窄依赖可以更有效地进行失效节点的恢复，即仅仅需又一次计算丢失RDD分区的父分区，并且不同节点之间可以并行计算。而对于一个宽依赖关系的Lineage图，单个节点失效可能导致这个RDD的祖先产生冗余计算开销。

    【补充】

    第一，窄依赖能够在某个计算节点上直接通过计算父RDD的某块数据计算得到子RDD相应的某块数据；宽依赖则要等到父RDD全部数据都计算完毕，而且父RDD的计算结果进行hash并传到相应节点上之后才干进行子RDD的计算。

• PageRank

    通过使用存储在HDFS上的54G大小的Wikipedia导出的数据，我们比較了Spark与Hadoop在PageRank计算方面的性能。PageRank算法通过10轮迭代处理了大约400万文章的链接图数据。图10显示了在30个节点上。基础的Spark处理速度是Hadoop的2.4倍。改进（对RDD进行Hash分区，来让迭代过程变得连贯）后速度提升到7.4倍，这些结果数据是随着节点扩展到60个而同步放大的。

    

watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQv/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center" alt="">

    我们也通过在Spark上实现Pregel来执行PageRank算法进行评估，在后面会提到。迭代次数和图10类似，可是耗时要多4秒。由于Pregel会在每一次的迭代过程中执行额外的操作来让节点“投票”决定是否结束当前job。

    【补充】Pregel：Google分布式图计算框架，主要用于图遍历、最短路径、PageRank算法等。

• Fault Recovery

    

    第5轮迭代结束时。交互时间大约为58秒。第6轮迭代时，一个节点被Kill掉，该节点上的任务都被终止，存储在其上的RDD的分区也丢失了。

Spark调度器调度这些任务在其它节点上又一次并行执行：又一次读取一直的输入数据并基于Lineage信息重建RDD。这使得迭代计算耗时添加到80秒。一旦丢失的RDD分区被重建，平均迭代时间又回落到58秒。

    须要注意的是，基于检查点的错误恢复机制（纯粹的检查点操作。并没有“血统”机制），恢复过程须要运行非常多次的迭代操作。

并且，系统须要通过网络将应用中多达100GB的数据进行复制。由于复制，Spark也许须要花费两倍的内存来完毕备份工作。

相反。我们样例中的RDD的“血统”图的存储只须要10KB大小的空间（存储在磁盘）。

• Behavior with Insufficient Memory（内存不足）

    到如今为止，我们能保证集群中的每一个节点都有足够的内存去缓存迭代过程中使用的RDD。那么问题来了，假设没有足够的内存来缓存一个作业的工作集，Spark又是怎样执行的呢？在实验中。我们通过在每一个节点上限制缓存RDD所须要的内存资源来对Spark进行配置，在不同的缓存配置条件下执行Logistic回归。结果如图12。我们能够看出，随着内存的减小，性能平缓地下降。

    

• User Applications Built with Spark

    In-Memory Analytics

    视频分发公司Conviva使用Spark极大地加快了一些数据分析报告的处理速度，之前是在Hadoop上执行的。比如，一个报告的处理相当于一系列的Hive[1]的查询处理过程。过程中对关于用户各种各样的统计数据进行了计算，这些查询作用在同样的数据子集上（数据满足用户提供的条件（过滤器）），可是在不同分组的字段上执行聚合操作（聚合函数）（SUM、AVG、COUNT
DISTINCT等）须要使用单独的MapReduce作业。

    通过运行Spark中的查询操作将共享的数据子集载入到一个RDD里面。该公司能够将报告的处理速度提升40倍。

在Hadoop集群上处理200G压缩数据生成的报告耗时20小时。而使用Spark基于96G内存的2个节点只需耗时30分钟就可以完毕，主要是由于Spark只将符合用户要求的信息的行列数据存储到了一个RDD里面，而不须要存储整个解压缩的文件。

    Traffic Modeling（城市交通建模）

    在Berkeley的Mobile Millennium项目中。研究人员基于一系列分散的汽车GPS监測数据，使用并行化学习算法来推算公路交通拥堵状况。数据来自10000个互联的公路线路网，还有600000个由汽车GPS装置採集到的点到点的旅行时间的样本数据，这些数据记录了汽车在两个地点之间行驶的时间（每一条路线的行驶时间可能跨多个公路线路网）。

使用一个交通模型。可以预计跨多个公路网行驶的耗时，由此，系统可以估算交通的拥堵状况。

研究人员使用Spark实现了一个可迭代的EM（最大期望）算法，算法迭代地反复运行map、reduceByKey操作。

应用从20个节点扩展到80个节点（每一个节点4核），如图13（a）所看到的。

    Twitter Spam（垃圾邮件）分类。Berkeley的Monarch项目使用Spark识别Twitter消息上的Spam链接。他们在Spark上实现了一个Logistic回归分类器。并使用分布式的reduceByKey操作对梯度向量并行求和。

图13（b）显示了基于50G数据子集的分类器的训练结果，整个数据集包含250000的URL、至少10^7个与网络相关的特征/维度以及每一个URL相应页面的内容属性。随着节点的添加，这并不像交通应用程序那样近似线性，主要是由于每轮迭代的固定通信代价较高。

    

watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQv/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center" alt="">

• Interactive Data Mining（交互式数据挖掘）

    为了展示Spark交互式查询大数据集的能力。我们在100个m2.4xlarge EC2实例（8核68G内存）上使用Spark分析两年间的1TB的Wikipedia页面浏览日志数据。在整个输入数据集上简单地查询例如以下内容以获取页面浏览总数：

    （1）所有页面；

    （2）页面的标题能精确匹配给定的关键词。

    （3）页面的标题能部分匹配给定的关键词。

    

Discussion

• Expressing Existing Programming Models

    RDD能够高效地实现眼下已经被提议独立的集群编程模型。所谓高效。就是RDDs不只能够实现跟这些模型一样的输出。还能够优化这些框架的性能。比方将特定的数据保存在内存。将RDD进行分区进而实现节点交互的优化。以及进行高效的数据恢复。这些能够使用RDDs进行实现的模型包含：

    MapReduce

    该模型能够使用Spark里面的flatMap和groupByKey算子来实现。或者是在有combiner的前提下用reduceByKey来实现。

    DryadLINQ

    该系统相对于MapReduce会提供更加广泛的算子，这些算子基本上都是Spark中的的总体计算转换算子，比方(map, groupByKey, join等

    SQL

    类似于DryadLINQ。SQL在记录集上运行数据并行查询的操作。

    Pregel

    Google的Pregel是专门用于图处理应用的模型，这个模型看起来跟其它面向数据集的程序模型非常不一样。在Pregel(基于Bulk Synchronous
Parallel。总体同步并行计算模型)中。程序由一系列超步（Superstep）协调迭代执行在每一个超步中，各个顶点执行用户函数，并更新对应的顶点状态。改变图的拓扑结构。然后向下一个超步的顶点集发送消息。

这样的模型可以描写叙述非常多图算法，包含最短路径，双边匹配和PageRank等。

    使用RDDs来完毕上述模型的主要原因是Pregel对全部定点的每一次迭代中使用了相似的用户程序。因此，我们能够将一个RDD中每一次迭代的定点状态信息进行存储。然后用该用户函数运行总体转换（flatmap）并将相关信息存储在新生成的RDD里面。我们能够将这个新生成的RDD和定点的状态进行join操作来进行定点的信息交换更新操作。

另一点非常重要，RDDs能够同意我们像Pregel一样将定点状态信息存储在内存中，这样就能够通过控制RDD的分区来优化集群节点之间的交互了。也能够支持部分的错误回复。

    Iterative MapReduce

    近期的好几个系统，包含Hadoop和Twister都提供了交互式的MapReduce模型。在该模型中。用户能够给系统一系列的job来迭代运行。此系统能够通过迭代连续地进行数据划分，且Twister能够将数据保存在内存其中。

    Batched Stream Processing

    研究人员近期提议了好几个增量的处理系统，这些系统都能够周期性地更新数据。比如，一个每15分钟对广告点击统计进行更新的应用，该应用能够将之前15分钟的状态和最新的log数据进行合并从而得到最新的状态。

这些系统运行和Dryad类似的总体操作。可是会在分布式文件系统中存储状态。将中间状态放入RDDs中能够加速处理过程。

    Explaining the Expressivity of RDDs

    为什么RDDs能够实现各种程序模型？由于附加在RDD上面的限制差点儿不会影响到大多数的并行应用。

特别地，尽管RDDs只只能通过总体的转换操作来创建，可是非常多的并行程序自然地会对记录採用同样的操作（适合总体操作的模式）。

相似地，RDDs的不可变性也不是什么障碍。由于我们能够创建多个RDD来来代表同样数据集的不同版本号。

其实，眼下使用文件系统的非常多的MapReduce应用都不支持文件的更新操作。

• Leveraging RDDs for Debugging

Related Work

• Cluster Programming Models

    眼下的集群模型分为好几类：

    （1）数据流模型。

像MapReduce、Dryad、Ciel那样拥有一系列处理数据的算子，可是是使用稳定的存储系统来进行数据的共享的。

    相对于那些稳定的存储系统，RDDs代表一种更加高效的数据共享抽象，由于RDDs避免了数据复制、I/O以及序列化的开销。

    （2）拥有高级程序接口的数据流系统。比方DryadLINQ、FlumeJava等，这些系统会提供相关的API使得用户能够通过map和join算子进行并行的数据获取。

然而，在这些系统中，并行的数据获取对象能够是磁盘中的文件或者是用于查询过程的临时的数据集。且系统会通过同样的算子来进行流水线式的数据处理（比方一系列的map操作），可是同样的算子的处理过程中，数据共享的效率太低。

    为了更加便利，我们在并行获取数据的过程中以正常的Spark API为基础。可是在口后背后会提供RDDs作为存储抽象。

    （3）为特定的须要数据共享的应用提供高级接口的系统。比如，Pregel支持迭代的图应用。而Twister和HaLoop是处理迭代式的MapReduce的。

然而，这些系统会为他们支持的计算模块提供数据共享。不会为用户提供普遍的数据共享的抽象。比如。用户无法使用Pregel或者Twister来将一个数据集载入到内存。并决定哪个数据查询的操作能够在上面运行。

    可是RDDs提供了一个分布式存储抽象而且能够支持哪些专门的系统无法支持的应用，比方交互式数据挖掘。

    （4）暴露了可分享的可变的状态的一些系统。比如。Piccolo让用户能够执行一些并行函数来读取或者更新一个分布式hash表的元素。

DSM(Distributed
shared memory )以及key-value存储（比方RAMCloud提供了相似的模型）也是类似的。

    RDDs与上述系统有两个方面的不同。第一。RDD基于算子map、sort和join等提供了高级编程接口，然而，Piccolo和DSM只只能读取以及更新hash表中的元素。第二，Piccolo和DSM系统通过检查点以及回滚操作来运行数据恢复，其开销要比RDDs的基于“血统”的错误容忍策略大非常多。

    Caching Systems

    Nectar可以通过识别带有程序分析的子表达式，跨DryadLINQ作业重用中间结果。这样的能力将会增加到基于RDD的系统。

可是Nectar并没有提供In-Memory缓存（使用一般的分布式文件系统）。也不可以让用户显式地控制应该缓存那个数据集，以及怎样对其进行分区。Ciel和FlumeJava相同可以记住任务结果，但不能提供In-Memory缓存并显式控制它的缓存方式。

    Ananthanarayanan系统通过利用数据存取的暂时性和空间本地性。进而添加了一个基于内存缓存的分布式文件系统。这样的方法提供了针对文件系统更快的数据存取速度，可是和共享中间结果的RDDs比起来效率还是不行，由于该系统还是须要应用将stage操作之间的结果写入文件系统进而共享数据。

   Lineage

    在科学计算和数据库领域。表示数据的Lineage和原始信息一直以来都是一个研究课题。对于一些应用。如须要解释结果以及同意被又一次生成、工作流中发现了bug或者数据集丢失须要又一次计算数据。RDD提供了一个并行的编程模型。在这个模型中使用的“血统”机制来表示是很easy的，因此它能够被用于容错。

    基于“血统”的恢复机制事实上与MapReduce以及Dryad的恢复机制是类似的，也会追踪tasks构成的DAG图的依赖关系。

然而在这些系统中，“血统”信息会在一个job结束之后丢失，最后还是须要採用在存储系统进行复制备份的方法来共享计算的数据。相反，RDDs用"血统"机制高效地将数据缓存至内存而不须要复制和I/O的开销。

    

   Relational Databases

    从概念上看，RDD类似于数据库中的视图，RDD的持久化过程类似于视图的物化。然而，像DSM系统一样，数据库同意典型地读写全部记录，通过操作日志和数据的日志来实现容错。还须要花费额外的开销来维护一致性。RDD编程模型粗粒度的转换模型能够避免这些开销。

Conclusion

    我们提出的RDD是一个高效的，通用的，具有错误容忍的应用于共享数据的集群应用的抽象。RDD广泛支持并行应用，还包含已经提出来的非常多专门的程序模型处理的迭代计算以及那些专门的程序模型不支持的新应用。

不像已经存在的通过数据复制进行容错处理的集群存储抽象。RDD基于粗粒度的转换操作提供了一些列的API，这些API可以基于“血统”机制进行高效率的数据恢复。

使用RDD的Spark系统实现处理迭代式作业的速度超过Hadoop大约20倍，并且还可以交互式查询数百GB的数据。