RDD之一：整体介绍

摘要

本文提出了分布式内存抽象的概念——弹性分布式数据集（RDD，Resilient Distributed Datasets），它具有像MapReduce等数据流模型的容错特性，而且容许开发人员在大型集群上执行基于内存的计算。现有的数据流系统对两种应用的处理并不高效：一是迭代式算法，这在图应用和机器学习领域很常见；二是交互式数据挖掘工具。这两种状况下，将数据保存在内存中可以极大地提升性能。为了有效地实现容错，RDD提供了一种高度受限的共享内存，即RDD是只读的，而且只能经过其余RDD上的批量操做来建立。尽管如此，RDD仍然足以表示不少类型的计算，包括MapReduce和专用的迭代编程模型（如Pregel）等。咱们实现的RDD在迭代计算方面比Hadoop快20多倍，同时还能够在5-7秒内交互式地查询1TB数据集。

1.引言

不管是工业界仍是学术界，都已经普遍使用高级集群编程模型来处理日益增加的数据，如MapReduce和Dryad。这些系统将分布式编程简化为自动提供位置感知性调度、容错以及负载均衡，使得大量用户可以在商用集群上分析超大数据集。

大多数现有的集群计算系统都是基于非循环的数据流模型。从稳定的物理存储（如分布式文件系统）中加载记录，记录被传入由一组肯定性操做构成的DAG，而后写回稳定存储。DAG数据流图可以在运行时自动实现任务调度和故障恢复。

尽管非循环数据流是一种很强大的抽象方法，但仍然有些应用没法使用这种方式描述。咱们就是针对这些不太适合非循环模型的应用，它们的特色是在多个并行操做之间重用工做数据集。这类应用包括：

（1）机器学习和图应用中经常使用的迭代算法（每一步对数据执行类似的函数）；

（2）交互式数据挖掘工具（用户反复查询一个数据子集）。基于数据流的框架并不明确支持工做集，因此须要将数据输出到磁盘，而后在每次查询时从新加载，这带来较大的开销。

咱们提出了一种分布式的内存抽象，称为弹性分布式数据集（RDD，Resilient Distributed Datasets）。它支持基于工做集的应用，同时具备数据流模型的特色：自动容错、位置感知调度和可伸缩性。RDD容许用户在执行多个查询时显式地将工做集缓存在内存中，后续的查询可以重用工做集，这极大地提高了查询速度。

RDD提供了一种高度受限的共享内存模型，即RDD是只读的记录分区的集合，只能经过在其余RDD执行肯定的转换操做（如map、join和group by）而建立，然而这些限制使得实现容错的开销很低。与分布式共享内存系统须要付出高昂代价的检查点和回滚机制不一样，RDD经过Lineage来重建丢失的分区：一个RDD中包含了如何从其余RDD衍生所必需的相关信息，从而不须要检查点操做就能够重构丢失的数据分区。尽管RDD不是一个通用的共享内存抽象，但却具有了良好的描述能力、可伸缩性和可靠性，但却可以普遍适用于数据并行类应用。

第一个指出非循环数据流存在不足的并不是是咱们，例如，Google的Pregel[21]，是一种专门用于迭代式图算法的编程模型；Twister[13]和HaLoop[8]，是两种典型的迭代式MapReduce模型。可是，对于一些特定类型的应用，这些系统提供了一个受限的通讯模型。相比之下，RDD则为基于工做集的应用提供了更为通用的抽象，用户能够对中间结果进行显式的命名和物化，控制其分区，还能执行用户选择的特定操做（而不是在运行时去循环执行一系列MapReduce步骤）。RDD能够用来描述Pregel、迭代式MapReduce，以及这两种模型没法描述的其余应用，如交互式数据挖掘工具（用户将数据集装入内存，而后执行ad-hoc查询）。

Spark是咱们实现的RDD系统，在咱们内部可以被用于开发多种并行应用。Spark采用Scala语言[5]实现，提供相似于DryadLINQ的集成语言编程接口[34]，使用户能够很是容易地编写并行任务。此外，随着Scala新版本解释器的完善，Spark还可以用于交互式查询大数据集。咱们相信Spark会是第一个可以使用有效、通用编程语言，并在集群上对大数据集进行交互式分析的系统。

咱们经过微基准和用户应用程序来评估RDD。实验代表，在处理迭代式应用上Spark比Hadoop快高达20多倍，计算数据分析类报表的性能提升了40多倍，同时可以在5-7秒的延时内交互式扫描1TB数据集。此外，咱们还在Spark之上实现了Pregel和HaLoop编程模型（包括其位置优化策略），以库的形式实现（分别使用了100和200行Scala代码）。最后，利用RDD内在的肯定性特性，咱们还建立了一种Spark调试工具rddbg，容许用户在任务期间利用Lineage重建RDD，而后像传统调试器那样从新执行任务。

本文首先在第2部分介绍了RDD的概念，而后第3部分描述Spark API，第4部分解释如何使用RDD表示几种并行应用（包括Pregel和HaLoop），第5部分讨论Spark中RDD的表示方法以及任务调度器，第6部分描述具体实现和rddbg，第7部分对RDD进行评估，第8部分给出了相关研究工做，最后第9部分总结。

2.弹性分布式数据集（RDD）

RDD的重要性：Spark生态圈中应用都是基于RDD构建（下图），这一点充分说明RDD的抽象足够通用，能够描述大多数应用场景。

本部分描述RDD和编程模型。首先讨论设计目标（2.1），而后定义RDD（2.2），讨论Spark的编程模型（2.3），并给出一个示例（2.4），最后对比RDD与分布式共享内存（2.5）。

RDD的特性：

一、RDD是Spark提供的核心抽象，全称为Resillient Distributed Dataset，即弹性分布式数据集。
二、RDD在抽象上来讲是一种元素集合，包含了数据。它是被分区的，分为多个分区，每一个分区分布在集群中的不一样节点上，从而让RDD中的数据能够被并行操做。（分布式数据集）
三、RDD一般经过Hadoop上的文件，即HDFS文件或者Hive表，来进行建立；有时也能够经过应用程序中的集合来建立。
四、RDD最重要的特性就是，提供了容错性，能够自动从节点失败中恢复过来。即若是某个节点上的RDD partition，由于节点故障，致使数据丢了，那么RDD会自动经过本身的数据来源从新计算该partition。这一切对使用者是透明的。
五、RDD的数据默认状况下存放在内存中的，可是在内存资源不足时，Spark会自动将RDD数据写入磁盘。（弹性）

一个RDD，在逻辑上，抽象地表明了一个HDFS文件。
可是，它其实是被分区得。分为多个分区。多个分区散落在Spark集群中，不一样的节点上。
好比说，RDD有90万数据。分为9个partition，9个分区。

如今，节点9出了些故障，致使partition9的数据丢失了。那么此时Spark会脆弱到直接报错，直接挂掉吗？不可能！！
RDD是有很强的容错性的，当它发现本身的数据丢失了之后，会自动从本身来源的数据进行重计算，从新获取本身这份数据，这一切对用户，都是彻底透明的。

RDD的每一个partition，在spark节点上存储时，默认都是放在内存中的。可是若是说内存放不下这么多数据时，好比每一个节点最多放5万数据，结果你每一个partition是10万数据。那么就会把partition中的部分数据写入磁盘上，进行保存。

2.1 目标和概述

咱们的目标是为基于工做集的应用（即多个并行操做重用中间结果的这类应用）提供抽象，同时保持MapReduce及其相关模型的优点特性：即自动容错、位置感知性调度和可伸缩性。RDD比数据流模型更易于编程，同时基于工做集的计算也具备良好的描述能力。

在这些特性中，最难实现的是容错性。通常来讲，分布式数据集的容错性有两种方式：即数据检查点和记录数据的更新。咱们面向的是大规模数据分析，数据检查点操做成本很高：须要经过数据中心的网络链接在机器之间复制庞大的数据集，而网络带宽每每比内存带宽低得多，同时还须要消耗更多的存储资源（在内存中复制数据能够减小须要缓存的数据量，而存储到磁盘则会拖慢应用程序）。因此，咱们选择记录更新的方式。可是，若是更新太多，那么记录更新成本也不低。所以，RDD只支持粗粒度转换，即在大量记录上执行的单个操做。将建立RDD的一系列转换记录下来（即Lineage），以便恢复丢失的分区。

虽然只支持粗粒度转换限制了编程模型，但咱们发现RDD仍然能够很好地适用于不少应用，特别是支持数据并行的批量分析应用，包括数据挖掘、机器学习、图算法等，由于这些程序一般都会在不少记录上执行相同的操做。RDD不太适合那些异步更新共享状态的应用，例如并行web爬行器。所以，咱们的目标是为大多数分析型应用提供有效的编程模型，而其余类型的应用交给专门的系统。

2.2 RDD抽象

RDD是只读的、分区记录的集合。RDD只能基于在稳定物理存储中的数据集和其余已有的RDD上执行肯定性操做来建立。这些肯定性操做称之为转换，如map、filter、groupBy、join（转换不是程开发人员在RDD上执行的操做）。

RDD不须要物化。RDD含有如何从其余RDD衍生（即计算）出本RDD的相关信息（即Lineage），据此能够从物理存储的数据计算出相应的RDD分区。

2.3 编程模型

 

Spark中，RDD被表示为对象，经过这些对象上的方法（或函数）调用转换。

定义RDD以后，程序员就能够在动做中使用RDD了。动做是向应用程序返回值，或向存储系统导出数据的那些操做，例如，count（返回RDD中的元素个数），collect（返回元素自己），save（将RDD输出到存储系统）。在Spark中，只有在动做第一次使用RDD时，才会计算RDD（即延迟计算）。这样在构建RDD的时候，运行时经过管道的方式传输多个转换。

程序员还能够从两个方面控制RDD，即缓存和分区。用户能够请求将RDD缓存，这样运行时将已经计算好的RDD分区存储起来，以加速后期的重用。缓存的RDD通常存储在内存中，但若是内存不够，能够写到磁盘上。

另外一方面，RDD还容许用户根据关键字（key）指定分区顺序，这是一个可选的功能。目前支持哈希分区和范围分区。例如，应用程序请求将两个RDD按照一样的哈希分区方式进行分区（将同一机器上具备相同关键字的记录放在一个分区），以加速它们之间的join操做。在Pregel和HaLoop中，屡次迭代之间采用一致性的分区置换策略进行优化，咱们一样也容许用户指定这种优化。

RDD操做类型—转换和动做

RDD的操做主要分两类：转换（transformation）和动做（action）。两类函数的主要区别是，转换接受RDD并返回RDD，而动做接受RDD可是返回非RDD。转换采用惰性调用机制，每一个RDD记录父RDD转换的方法，这种调用链表称之为血缘（lineage）；而动做调用会直接计算。

采用惰性调用，经过血缘链接的RDD操做能够管道化（pipeline），管道化的操做能够直接在单节点完成，避免屡次转换操做之间数据同步的等待。

使用血缘串联的操做能够保持每次计算相对简单，而不用担忧有过多的中间数据，由于这些血缘操做都管道化了，这样也保证了逻辑的单一性，而不用像MapReduce那样，为了竟可能的减小map reduce过程，在单个map reduce中写入过多复杂的逻辑。

RDD使用模式

   

RDD使用具备通常的模式，能够抽象为下面的几步

1. 加载外部数据，建立RDD对象
2. 使用转换（如filter），建立新的RDD对象
3. 缓存须要重用的RDD
4. 使用动做（如count），启动并行计算

   

RDD高效的策略

Spark官方提供的数据是RDD在某些场景下，计算效率是Hadoop的20X。这个数据是否有水分，咱们先不追究，可是RDD效率高的由必定机制保证的：

1. RDD数据只读，不可修改。若是须要修改数据，必须从父RDD转换（transformation）到子RDD。因此，在容错策略中， RDD没有数据冗余，而是经过RDD父子依赖（血缘）关系进行重算实现容错。
2. RDD数据在内存中，多个RDD操做之间，数据不用落地到磁盘上，避免没必要要的I/O操做。
3. RDD存放的数据能够是java对象，因此避免的没必要要的对象序列化和反序列化。

总而言之，RDD高效的主要因素是尽可能避免没必要要的操做和牺牲数据的操做精度，用来提升计算效率。

Spark使用技巧

RDD基本函数扩展

   

RDD虽然提供了不少函数，可是毕竟仍是有限的，有时候须要扩展，自定义新的RDD的函数。在spark中，能够经过隐式转换，轻松实现对RDD扩展。画像开发过程当中，平凡的会使用rollup操做（相似HIVE中的rollup），计算多个级别的聚合数据。下面是具体实，

/** * 扩展spark rdd,为rdd提供rollup方法 */ implicit class RollupRDD[T: ClassTag](rdd: RDD[(Array[String], T)]) extends Serializable {   /** * 相似Sql中的rollup操做 * * @param aggregate 聚合函数 * @param keyPlaceHold key占位符，默认采用FaceConf.STAT_SUMMARY * @param isCache，确认是否缓存数据 * @return 返回聚合后的数据 */ def rollup[U: ClassTag]( aggregate: Iterable[T] => U, keyPlaceHold: String = FaceConf.STAT_SUMMARY, isCache: Boolean = true): RDD[(Array[String], U)] = {   if (rdd.take(1).isEmpty) { return rdd.map(x => (Array[String](), aggregate(Array[T](x._2)))) }   if (isCache) { rdd.cache // 提升计算效率 } val totalKeyCount = rdd.first._1.size val result = { 1 to totalKeyCount }.par.map(untilKeyIndex => { // 并行计算 rdd.map(row => { val combineKey = row._1.slice(0, untilKeyIndex).mkString(FaceConf.KEY_SEP) // 组合key (combineKey, row._2) }).groupByKey.map(row => { // 聚合计算 val oldKeyList = row._1.split(FaceConf.KEY_SEP) val newKeyList = oldKeyList ++ Array.fill(totalKeyCount - oldKeyList.size) { keyPlaceHold } (newKeyList, aggregate(row._2)) }) }).reduce(_ ++ _) // 聚合结果   result }   }

上面代码声明了一个隐式类，具备一个成员变量rdd，类型是RDD[(Array[String], T)]，那么若是应用代码中出现了任何这样的rdd对象，而且import当前的隐式转换，那么编译器就会将这个rdd当作上面的隐式类的对象，也就可使用rollup函数，和通常的map，filter方法同样。

   

   

RDD操做闭包外部变量原则

   

RDD相关操做都须要传入自定义闭包函数（closure），若是这个函数须要访问外部变量，那么须要遵循必定的规则，不然会抛出运行时异常。闭包函数传入到节点时，须要通过下面的步骤：

1. 驱动程序，经过反射，运行时找到闭包访问的全部变量，并封成一个对象，而后序列化该对象
2. 将序列化后的对象经过网络传输到worker节点
3. worker节点反序列化闭包对象
4. worker节点执行闭包函数，

注意：外部变量在闭包内的修改不会被反馈到驱动程序。

简而言之，就是经过网络，传递函数，而后执行。因此，被传递的变量必须能够序列化，不然传递失败。本地执行时，仍然会执行上面四步。

   

广播机制也能够作到这一点，可是频繁的使用广播会使代码不够简洁，并且广播设计的初衷是将较大数据缓存到节点上，避免屡次数据传输，提升计算效率，而不是用于进行外部变量访问。

   

   

RDD数据同步

   

RDD目前提供两个数据同步的方法：广播和累计器。

   

广播 broadcast

前面提到过，广播能够将变量发送到闭包中，被闭包使用。可是，广播还有一个做用是同步较大数据。好比你有一个IP库，可能有几G，在map操做中，依赖这个ip库。那么，能够经过广播将这个ip库传到闭包中，被并行的任务应用。广播经过两个方面提升数据共享效率：1，集群中每一个节点（物理机器）只有一个副本，默认的闭包是每一个任务一个副本；2，广播传输是经过BT下载模式实现的，也就是P2P下载，在集群多的状况下，能够极大的提升数据传输速率。广播变量修改后，不会反馈到其余节点。

   

累加器 Accumulator

累加器是一个write-only的变量，用于累加各个任务中的状态，只有在驱动程序中，才能访问累加器。并且，截止到1.2版本，累加器有一个已知的缺陷，在action操做中，n个元素的RDD能够确保累加器只累加n次，可是在transformation时，spark不确保，也就是累加器可能出现n+1次累加。

   

目前RDD提供的同步机制粒度太粗，尤为是转换操做中变量状态不能同步，因此RDD没法作复杂的具备状态的事务操做。不过，RDD的使命是提供一个通用的并行计算框架，估计永远也不会提供细粒度的数据同步机制，由于这与其设计的初衷是违背的。

   

RDD优化技巧

   

RDD缓存

须要使用屡次的数据须要cache，不然会进行没必要要的重复操做。举个例子

val data = … // read from tdw println(data.filter(_.contains("error")).count) println(data.filter(_.contains("warning")).count)

上面三段代码中，data变量会加载两次，高效的作法是在data加载完后，马上持久化到内存中，以下

val data = … // read from tdw data.cache println(data.filter(_.contains("error")).count) println(data.filter(_.contains("warning")).count)

这样，data在第一加载后，就被缓存到内存中，后面两次操做均直接使用内存中的数据。

   

转换并行化

RDD的转换操做时并行化计算的，可是多个RDD的转换一样是能够并行的，参考以下

val dataList:Array[RDD[Int]] = … val sumList = data.list.map(_.map(_.sum))

上面的例子中，第一个map是便利Array变量，串行的计算每一个RDD中的每行的sum。因为每一个RDD之间计算是没有逻辑联系的，因此理论上是能够将RDD的计算并行化的，在scala中能够轻松试下，以下

val dataList:Array[RDD[Int]] = … val sumList = data.list.par.map(_.map(_.sum))

注意红色代码。

   

减小shuffle网络传输

通常而言，网络I/O开销是很大的，减小网络开销，能够显著加快计算效率。任意两个RDD的shuffle操做（join等）的大体过程以下，

用户数据userData和事件events数据经过用户id链接，那么会在网络中传到另一个节点，这个过程当中，有两个网络传输过程。Spark的默认是完成这两个过程。可是，若是你多告诉spark一些信息，spark能够优化，只执行一个网络传输。能够经过使用、HashPartition，在userData"本地"先分区，而后要求events直接shuffle到userData的节点上，那么就减小了一部分网络传输，减小后的效果以下，

虚线部分都是在本地完成的，没有网络传输。在数据加载时，就按照key进行partition，这样能够经一部的减小本地的HashPartition的过程，示例代码以下

val userData = sc.sequenceFile[UserID, UserInfo]("hdfs://…") .partitionBy(new HashPartitioner(100)) // Create 100 partitions .persist()

注意，上面必定要persist，不然会重复计算屡次。100用来指定并行数量。

   

Spark其余

   

Spark开发模式

   

因为spark应用程序是须要在部署到集群上运行的，致使本地调试比较麻烦，因此通过这段时间的经验累积，总结了一套开发流程，目的是为了尽量的提升开发调试效率，同时保证开发质量。固然，这套流程可能也不是最优的，后面须要持续改进。

整个流程比较清楚，这里主要谈谈为何须要单元测试。公司内的大多数项目，通常不提倡单元测试，并且因为项目进度压力，开发人员会很是抵触单元测试，由于会花费"额外"的精力。Bug这东西不会由于项目赶进度而消失，并且刚好相反，可能由于赶进度，而高于平均水平。因此，若是不花时间进行单元测试，那么会花一样多，甚至更多的时间调试。不少时候，每每一些很小的bug，却致使你花了很长时间去调试，而这些bug，刚好是很容易在单元测试中发现的。并且，单元测试还能够带来两个额外的好处：1）API使用范例；2）回归测试。因此，仍是单元测试吧，这是一笔投资，并且ROI还挺高！不过凡事须要掌握分寸，单元测试应该根据项目紧迫程度调整粒度，作到有所为，有所不为。

 

Spark其余功能

   

前面提到了spark生态圈，spark除了核心的RDD，还提供了之上的几个很使用的应用：

1. Spark SQL: 相似hive，使用rdd实现sql查询
2. Spark Streaming: 流式计算，提供实时计算功能，相似storm
3. MLLib：机器学习库，提供经常使用分类，聚类，回归，交叉检验等机器学习算法并行实现。
4. GraphX：图计算框架，实现了基本的图计算功能，经常使用图算法和pregel图编程框架。

   

后面须要继续学习和使用上面的功能，尤为是与数据挖掘强相关的MLLib。

2.4 示例：控制台日志挖掘

本部分咱们经过一个具体示例来阐述RDD。假定有一个大型网站出错，操做员想要检查Hadoop文件系统（HDFS）中的日志文件（TB级大小）来找出缘由。经过使用Spark，操做员只需将日志中的错误信息装载到一组节点的内存中，而后执行交互式查询。首先，须要在Spark解释器中输入以下Scala命令：

1	lines = spark.textFile("hdfs://...")

2	errors = lines.filter(_.startsWith("ERROR"))

3	errors.cache()

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

这时集群尚未开始执行任何任务。可是，用户已经能够在这个RDD上执行对应的动做，例如统计错误消息的数目：

1	errors.count()

用户还能够在RDD上执行更多的转换操做，并使用转换结果，如：

1	// Count errors mentioning MySQL:

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

3	// Return the time fields of errors mentioning

4	// HDFS as an array (assuming time is field

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

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

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

8	.collect()

使用errors的第一个action运行之后，Spark会把errors的分区缓存在内存中，极大地加快了后续计算速度。注意，最初的RDD lines不会被缓存。由于错误信息可能只占原数据集的很小一部分（小到足以放入内存）。
最后，为了说明模型的容错性，图1给出了第3个查询的Lineage图。在lines RDD上执行filter操做，获得errors，而后再filter、map后获得新的RDD，在这个RDD上执行collect操做。Spark调度器以流水线的方式执行后两个转换，向拥有errors分区缓存的节点发送一组任务。此外，若是某个errors分区丢失，Spark只在相应的lines分区上执行filter操做来重建该errors分区。

图1 示例中第三个查询的Lineage图。（方框表示RDD，箭头表示转换）

2.5 RDD与分布式共享内存

为了进一步理解RDD是一种分布式的内存抽象，表1列出了RDD与分布式共享内存（DSM，Distributed Shared Memory）[24]的对比。在DSM系统中，应用能够向全局地址空间的任意位置进行读写操做。（注意这里的DSM，不只指传统的共享内存系统，还包括那些经过分布式哈希表或分布式文件系统进行数据共享的系统，好比Piccolo[28]）DSM是一种通用的抽象，但这种通用性同时也使得在商用集群上实现有效的容错性更加困难。

RDD与DSM主要区别在于，不只能够经过批量转换建立（即“写”）RDD，还能够对任意内存位置读写。也就是说，RDD限制应用执行批量写操做，这样有利于实现有效的容错。特别地，RDD没有检查点开销，由于可使用Lineage来恢复RDD。并且，失效时只须要从新计算丢失的那些RDD分区，能够在不一样节点上并行执行，而不须要回滚整个程序。

表1 RDD与DSM对比

对比项目	RDD	分布式共享内存（DSM）
读	批量或细粒度操做	细粒度操做
写	批量转换操做	细粒度操做
一致性	不重要（RDD是不可更改的）	取决于应用程序或运行时
容错性	细粒度，低开销（使用Lineage）	须要检查点操做和程序回滚
落后任务的处理	任务备份	很难处理
任务安排	基于数据存放的位置自动实现	取决于应用程序（经过运行时实现透明性）
若是内存不够	与已有的数据流系统相似	性能较差（交换？）

注意，经过备份任务的拷贝，RDD还能够处理落后任务（即运行很慢的节点），这点与MapReduce[12]相似。而DSM则难以实现备份任务，由于任务及其副本都须要读写同一个内存位置。

与DSM相比，RDD模型有两个好处。第一，对于RDD中的批量操做，运行时将根据数据存放的位置来调度任务，从而提升性能。第二，对于基于扫描的操做，若是内存不足以缓存整个RDD，就进行部分缓存。把内存放不下的分区存储到磁盘上，此时性能与现有的数据流系统差很少。

最后看一下读操做的粒度。RDD上的不少动做（如count和collect）都是批量读操做，即扫描整个数据集，能够将任务分配到距离数据最近的节点上。同时，RDD也支持细粒度操做，即在哈希或范围分区的RDD上执行关键字查找。

3. Spark编程接口

Spark用Scala[5]语言实现了RDD的API。Scala是一种基于JVM的静态类型、函数式、面向对象的语言。咱们选择Scala是由于它简洁（特别适合交互式使用）、有效（由于是静态类型）。可是，RDD抽象并不局限于函数式语言，也可使用其余语言来实现RDD，好比像Hadoop[2]那样用类表示用户函数。

要使用Spark，开发者须要编写一个driver程序，链接到集群以运行Worker，如图2所示。Driver定义了一个或多个RDD，并调用RDD上的动做。Worker是长时间运行的进程，将RDD分区以Java对象的形式缓存在内存中。

图2 Spark的运行时。用户的driver程序启动多个worker，worker从分布式文件系统中读取数据块，并将计算后的RDD分区缓存在内存中。

再看看2.4中的例子，用户执行RDD操做时会提供参数，好比map传递一个闭包（closure，函数式编程中的概念）。Scala将闭包表示为Java对象，若是传递的参数是闭包，则这些对象被序列化，经过网络传输到其余节点上进行装载。Scala将闭包内的变量保存为Java对象的字段。例如，var x = 5; rdd.map(_ + x) 这段代码将RDD中的每一个元素加5。总的来讲，Spark的语言集成相似于DryadLINQ。

RDD自己是静态类型对象，由参数指定其元素类型。例如，RDD[int]是一个整型RDD。不过，咱们举的例子几乎都省略了这个类型参数，由于Scala支持类型推断。

虽然在概念上使用Scala实现RDD很简单，但仍是要处理一些Scala闭包对象的反射问题。如何经过Scala解释器来使用Spark还须要更多工做，这点咱们将在第6部分讨论。无论怎样，咱们都不须要修改Scala编译器。

3.1 Spark中的RDD操做

表2列出了Spark中的RDD转换和动做。每一个操做都给出了标识，其中方括号表示类型参数。前面说过转换是延迟操做，用于定义新的RDD；而动做启动计算操做，并向用户程序返回值或向外部存储写数据。

表3 Spark中支持的RDD转换和动做

转换

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)] 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] 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

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

除了这些操做之外，用户还能够请求将RDD缓存起来。并且，用户还能够经过Partitioner类获取RDD的分区顺序，而后将另外一个RDD按照一样的方式分区。有些操做会自动产生一个哈希或范围分区的RDD，像groupByKey，reduceByKey和sort等。

4. 应用程序示例

如今咱们讲述如何使用RDD表示几种基于数据并行的应用。首先讨论一些迭代式机器学习应用（4.1），而后看看如何使用RDD描述几种已有的集群编程模型，即MapReduce（4.2），Pregel（4.3），和Hadoop（4.4）。最后讨论一下RDD不适合哪些应用（4.5）。

4.1 迭代式机器学习

不少机器学习算法都具备迭代特性，运行迭代优化方法来优化某个目标函数，例如梯度降低方法。若是这些算法的工做集可以放入内存，将极大地加速程序运行。并且，这些算法一般采用批量操做，例如映射和求和，这样更容易使用RDD来表示。

例以下面的程序是逻辑回归[15]的实现。逻辑回归是一种常见的分类算法，即寻找一个最佳分割两组点（即垃圾邮件和非垃圾邮件）的超平面w。算法采用梯度降低的方法：开始时w为随机值，在每一次迭代的过程当中，对w的函数求和，而后朝着优化的方向移动w。

1	val points = spark.textFile(...)

2	.map(parsePoint).persist()

3	var w = // random initial vector

4	for (i <- 1 to ITERATIONS) {

5	val gradient = points.map{ p =>

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

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

8	w -= gradient

9

}

首先定义一个名为points的缓存RDD，这是在文本文件上执行map转换以后获得的，即将每一个文本行解析为一个Point对象。而后在points上反复执行map和reduce操做，每次迭代时经过对当前w的函数进行求和来计算梯度。7.1小节咱们将看到这种在内存中缓存points的方式，比每次迭代都从磁盘文件装载数据并进行解析要快得多。

已经在Spark中实现的迭代式机器学习算法还有：kmeans（像逻辑回归同样每次迭代时执行一对map和reduce操做），指望最大化算法（EM，两个不一样的map/reduce步骤交替执行），交替最小二乘矩阵分解和协同过滤算法。Chu等人提出迭代式MapReduce也能够用来实现经常使用的学习算法[11]。

4.2 使用RDD实现MapReduce

MapReduce模型[12]很容易使用RDD进行描述。假设有一个输入数据集（其元素类型为T），和两个函数myMap: T => List[(Ki, Vi)] 和 myReduce: (Ki; List[Vi]) ) List[R]，代码以下：

1	data.flatMap(myMap)

2	.groupByKey()

3	.map((k, vs) => myReduce(k, vs))

若是任务包含combiner，则相应的代码为：

1	data.flatMap(myMap)

2	.reduceByKey(myCombiner)

3	.map((k, v) => myReduce(k, v))

ReduceByKey操做在mapper节点上执行部分汇集，与MapReduce的combiner相似。

4.3 使用RDD实现Pregel

Pregel[21]是面向图算法的基于BSP范式[32]的编程模型。程序由一系列超步（Superstep）协调迭代运行。在每一个超步中，各个顶点执行用户函数，并更新相应的顶点状态，变异图拓扑，而后向下一个超步的顶点集发送消息。这种模型可以描述不少图算法，包括最短路径，双边匹配和PageRank等。

以PageRank为例介绍一下Pregel的实现。当前PageRank[7]记为r，顶点表示状态。在每一个超步中，各个顶点向其全部邻居发送贡献值r/n，这里n是邻居的数目。下一个超步开始时，每一个顶点将其分值（rank）更新为 α/N + (1 - α) * Σci，这里的求和是各个顶点收到的全部贡献值的和，N是顶点的总数。

Pregel将输入的图划分到各个worker上，并存储在其内存中。在每一个超步中，各个worker经过一种相似MapReduce的Shuffle操做交换消息。

Pregel的通讯模式能够用RDD来描述，如图3。主要思想是：将每一个超步中的顶点状态和要发送的消息存储为RDD，而后根据顶点ID分组，进行Shuffle通讯（即cogroup操做）。而后对每一个顶点ID上的状态和消息应用用户函数（即mapValues操做），产生一个新的RDD，即(VertexID, (NewState, OutgoingMessages))。而后执行map操做分离出下一次迭代的顶点状态和消息（即mapValues和flatMap操做）。代码以下：

1	val vertices = // RDD of (ID, State) pairs

2	val messages = // RDD of (ID, Message) pairs

3	val grouped = vertices.cogroup(messages)

4	val newData = grouped.mapValues {

5	(vert, msgs) => userFunc(vert, msgs)

6	// returns (newState, outgoingMsgs)

7

}.cache()

8	val newVerts = newData.mapValues((v,ms) => v)

9	val newMsgs = newData.flatMap((id,(v,ms)) => ms)

图3 使用RDD实现Pregel时，一步迭代的数据流。（方框表示RDD，箭头表示转换）
须要注意的是，这种实现方法中，RDD grouped，newData和newVerts的分区方法与输入RDD vertices同样。因此，顶点状态一直存在于它们开始执行的机器上，这跟原Pregel同样，这样就减小了通讯成本。由于cogroup和mapValues保持了与输入RDD相同的分区方法，因此分区是自动进行的。

完整的Pregel编程模型还包括其余工具，好比combiner，附录A讨论了它们的实现。下面将讨论Pregel的容错性，以及如何在实现相同容错性的同时减小须要执行检查点操做的数据量。

咱们差很少用了100行Scala代码在Spark上实现了一个类Pregel的API。7.2小节将使用PageRank算法评估它的性能。

4.3.1 Pregel容错

当前，Pregel基于检查点机制来为顶点状态及其消息实现容错[21]。然而做者是这样描述的：经过在其它的节点上记录已发消息日志，而后单独重建丢失的分区，只须要恢复局部数据便可。上面提到这两种方式，RDD都可以很好地支持。

经过4.3小节的实现，Spark老是可以基于Lineage实现顶点和消息RDD的重建，可是因为过长的Lineage链，恢复可能会付出高昂的代价。由于迭代RDD依赖于上一个RDD，对于部分分区来讲，节点故障可能会致使这些分区状态的全部迭代版本丢失，这就要求使用一种“级联-从新执行”[20]的方式去依次重建每个丢失的分区。为了不这个问题，用户能够周期性地在顶点和消息RDD上执行save操做，将状态信息保存到持久存储中。而后，Spark可以在失败的时候自动地从新计算这些丢失的分区（而不是回滚整个程序）。

最后，咱们意识到，RDD也可以实现检查点数据的reduce操做，这要求经过一种高效的检查点方案来表达检查点数据。在不少Pregel做业中，顶点状态都包括可变与不可变的组件，例如，在PageRank中，与一个顶点相邻的顶点列表是不可变的，可是它们的排名是可变的，在这种状况下，咱们可使用一个来自可变数据的单独RDD来替换不可变RDD，基于这样一个较短的Lineage链，检查点仅仅是可变状态，图4解释了这种方式。

图4 通过优化的Pregel使用RDD的数据流。可变状态RDD必须设置检查点，不可变状态才可被快速重建。
在PageRank中，不可变状态（相邻顶点列表）远大于可变状态（浮点值），因此这种方式可以极大地下降开销。

4.4 使用RDD实现HaLoop

HaLoop[8]是Hadoop的一个扩展版本，它可以改善具备迭代特性的MapReduce程序的性能。基于HaLoop编程模型的应用，使用reduce阶段的输出做为map阶段下一轮迭代的输入。它的循环感知任务调度器可以保证，在每一轮迭代中处理同一个分区数据的连续map和reduce任务，必定可以在同一台物理机上执行。确保迭代间locality特性，reduce数据在物理节点之间传输，而且容许数据缓存在本地磁盘而可以被后续迭代重用。

使用RDD来优化HaLoop，咱们在Spark上实现了一个相似HaLoop的API，这个库只使用了200行Scala代码。经过partitionBy可以保证跨迭代的分区的一致性，每个阶段的输入和输出被缓存以用于后续迭代。

4.5 不适合使用RDD的应用

在2.1节咱们讨论过，RDD适用于具备批量转换需求的应用，而且相同的操做做用于数据集的每个元素上。在这种状况下，RDD可以记住每一个转换操做，对应于Lineage图中的一个步骤，恢复丢失分区数据时不须要写日志记录大量数据。RDD不适合那些经过异步细粒度地更新来共享状态的应用，例如Web应用中的存储系统，或者增量抓取和索引Web数据的系统，这样的应用更适合使用一些传统的方法，例如数据库、RAMCloud[26]、Percolator[27]和Piccolo[28]。咱们的目标是，面向批量分析应用的这类特定系统，提供一种高效的编程模型，而不是一些异步应用程序。

5. RDD的描述及做业调度

咱们但愿在不修改调度器的前提下，支持RDD上的各类转换操做，同时可以从这些转换获取Lineage信息。为此，咱们为RDD设计了一组小型通用的内部接口。

简单地说，每一个RDD都包含：（1）一组RDD分区（partition，即数据集的原子组成部分）；（2）对父RDD的一组依赖，这些依赖描述了RDD的Lineage；（3）一个函数，即在父RDD上执行何种计算；（4）元数据，描述分区模式和数据存放的位置。例如，一个表示HDFS文件的RDD包含：各个数据块的一个分区，并知道各个数据块放在哪些节点上。并且这个RDD上的map操做结果也具备一样的分区，map函数是在父数据上执行的。表3总结了RDD的内部接口。

表3 Spark中RDD的内部接口

操做	含义
partitions()	返回一组Partition对象
preferredLocations(p)	根据数据存放的位置，返回分区p在哪些节点访问更快
dependencies()	返回一组依赖
iterator(p, parentIters)	按照父分区的迭代器，逐个计算分区p的元素
partitioner()	返回RDD是否hash/range分区的元数据信息

设计接口的一个关键问题就是，如何表示RDD之间的依赖。咱们发现RDD之间的依赖关系能够分为两类，即：（1）窄依赖（narrow dependencies）：子RDD的每一个分区依赖于常数个父分区（即与数据规模无关）；（2）宽依赖（wide dependencies）：子RDD的每一个分区依赖于全部父RDD分区。例如，map产生窄依赖，而join则是宽依赖（除非父RDD被哈希分区）。另外一个例子见图5。

图5 窄依赖和宽依赖的例子。（方框表示RDD，实心矩形表示分区）
区分这两种依赖颇有用。首先，窄依赖容许在一个集群节点上以流水线的方式（pipeline）计算全部父分区。例如，逐个元素地执行map、而后filter操做；而宽依赖则须要首先计算好全部父分区数据，而后在节点之间进行Shuffle，这与MapReduce相似。第二，窄依赖可以更有效地进行失效节点的恢复，即只需从新计算丢失RDD分区的父分区，并且不一样节点之间能够并行计算；而对于一个宽依赖关系的Lineage图，单个节点失效可能致使这个RDD的全部祖先丢失部分分区，于是须要总体从新计算。

经过RDD接口，Spark只须要不超过20行代码实现即可以实现大多数转换。5.1小节给出了例子，而后咱们讨论了怎样使用RDD接口进行调度（5.2），最后讨论一下基于RDD的程序什么时候须要数据检查点操做（5.3）。

5.1 RDD实现举例

HDFS文件：目前为止咱们给的例子中输入RDD都是HDFS文件，对这些RDD能够执行：partitions操做返回各个数据块的一个分区（每一个Partition对象中保存数据块的偏移），preferredLocations操做返回数据块所在的节点列表，iterator操做对数据块进行读取。

map：任何RDD上均可以执行map操做，返回一个MappedRDD对象。该操做传递一个函数参数给map，对父RDD上的记录按照iterator的方式执行这个函数，并返回一组符合条件的父RDD分区及其位置。

union：在两个RDD上执行union操做，返回两个父RDD分区的并集。经过相应父RDD上的窄依赖关系计算每一个子RDD分区（注意union操做不会过滤重复值，至关于SQL中的UNION ALL）。

sample：抽样与映射相似，可是sample操做中，RDD须要存储一个随机数产生器的种子，这样每一个分区可以肯定哪些父RDD记录被抽样。

join：对两个RDD执行join操做可能产生窄依赖（若是这两个RDD拥有相同的哈希分区或范围分区），多是宽依赖，也可能两种依赖都有（好比一个父RDD有分区，而另外一父RDD没有）。

5.2 Spark任务调度器

调度器根据RDD的结构信息为每一个动做肯定有效的执行计划。调度器的接口是runJob函数，参数为RDD及其分区集，和一个RDD分区上的函数。该接口足以表示Spark中的全部动做（即count、collect、save等）。

总的来讲，咱们的调度器跟Dryad相似，但咱们还考虑了哪些RDD分区是缓存在内存中的。调度器根据目标RDD的Lineage图建立一个由stage构成的无回路有向图（DAG）。每一个stage内部尽量多地包含一组具备窄依赖关系的转换，并将它们流水线并行化（pipeline）。stage的边界有两种状况：一是宽依赖上的Shuffle操做；二是已缓存分区，它能够缩短父RDD的计算过程。例如图6。父RDD完成计算后，能够在stage内启动一组任务计算丢失的分区。

图6 Spark怎样划分任务阶段（stage）的例子。实线方框表示RDD，实心矩形表示分区（黑色表示该分区被缓存）。要在RDD G上执行一个动做，调度器根据宽依赖建立一组stage，并在每一个stage内部将具备窄依赖的转换流水线化（pipeline）。 本例不用再执行stage 1，由于B已经存在于缓存中了，因此只须要运行2和3。

调度器根据数据存放的位置分配任务，以最小化通讯开销。若是某个任务须要处理一个已缓存分区，则直接将任务分配给拥有这个分区的节点。不然，若是须要处理的分区位于多个可能的位置（例如，由HDFS的数据存放位置决定），则将任务分配给这一组节点。

对于宽依赖（例如须要Shuffle的依赖），目前的实现方式是，在拥有父分区的节点上将中间结果物化，简化容错处理，这跟MapReduce中物化map输出很像。

若是某个任务失效，只要stage中的父RDD分区可用，则只需在另外一个节点上从新运行这个任务便可。若是某些stage不可用（例如，Shuffle时某个map输出丢失），则须要从新提交这个stage中的全部任务来计算丢失的分区。

最后，lookup动做容许用户从一个哈希或范围分区的RDD上，根据关键字读取一个数据元素。这里有一个设计问题。Driver程序调用lookup时，只须要使用当前调度器接口计算关键字所在的那个分区。固然任务也能够在集群上调用lookup，这时能够将RDD视为一个大的分布式哈希表。这种状况下，任务和被查询的RDD之间的并无明确的依赖关系（由于worker执行的是lookup），若是全部节点上都没有相应的缓存分区，那么任务须要告诉调度器计算哪些RDD来完成查找操做。

5.3 检查点

尽管RDD中的Lineage信息能够用来故障恢复，但对于那些Lineage链较长的RDD来讲，这种恢复可能很耗时。例如4.3小节中的Pregel任务，每次迭代的顶点状态和消息都跟前一次迭代有关，因此Lineage链很长。若是将Lineage链存到物理存储中，再按期对RDD执行检查点操做就颇有效。

通常来讲，Lineage链较长、宽依赖的RDD须要采用检查点机制。这种状况下，集群的节点故障可能致使每一个父RDD的数据块丢失，所以须要所有从新计算[20]。将窄依赖的RDD数据存到物理存储中能够实现优化，例如前面4.1小节逻辑回归的例子，将数据点和不变的顶点状态存储起来，就再也不须要检查点操做。

当前Spark版本提供检查点API，但由用户决定是否须要执行检查点操做。从此咱们将实现自动检查点，根据成本效益分析肯定RDD Lineage图中的最佳检查点位置。

值得注意的是，由于RDD是只读的，因此不须要任何一致性维护（例如写复制策略，分布式快照或者程序暂停等）带来的开销，后台执行检查点操做。

咱们使用10000行Scala代码实现了Spark。系统可使用任何Hadoop数据源（如HDFS，Hbase）做为输入，这样很容易与Hadoop环境集成。Spark以库的形式实现，不须要修改Scala编译器。

这里讨论关于实现的三方面问题：（1）修改Scala解释器，容许交互模式使用Spark（6.1）；（2）缓存管理（6.2）；（3）调试工具rddbg（6.3）。

6. 实现

6.1 解释器的集成

像Ruby和Python同样，Scala也有一个交互式shell。基于内存的数据能够实现低延时，咱们但愿容许用户从解释器交互式地运行Spark，从而在大数据集上实现大规模并行数据挖掘。

Scala解释器一般根据将用户输入的代码行，来对类进行编译，接着装载到JVM中，而后调用类的函数。这个类是一个包含输入行变量或函数的单例对象，并在一个初始化函数中运行这行代码。例如，若是用户输入代码var x = 5，接着又输入println(x)，则解释器会定义一个包含x的Line1类，并将第2行编译为println(Line1.getInstance().x)。

在Spark中咱们对解释器作了两点改动：

1. 类传输：解释器可以支持基于HTTP传输类字节码，这样worker节点就能获取输入每行代码对应的类的字节码。
2. 改进的代码生成逻辑：一般每行上建立的单态对象经过对应类上的静态方法进行访问。也就是说，若是要序列化一个闭包，它引用了前面代码行中变量，好比上面的例子Line1.x，Java不会根据对象关系传输包含x的Line1实例。因此worker节点不会收到x。咱们将这种代码生成逻辑改成直接引用各个行对象的实例。图7说明了解释器如何将用户输入的一组代码行解释为Java对象。

图7 Spark解释器如何将用户输入的两行代码解释为Java对象
Spark解释器便于跟踪处理大量对象关系引用，而且便利了HDFS数据集的研究。咱们计划以Spark解释器为基础，开发提供高级数据分析语言支持的交互式工具，好比相似SQL和Matlab。

6.2 缓存管理

Worker节点将RDD分区以Java对象的形式缓存在内存中。因为大部分操做是基于扫描的，采起RDD级的LRU（最近最少使用）替换策略（即不会为了装载一个RDD分区而将同一RDD的其余分区替换出去）。目前这种简单的策略适合大多数用户应用。另外，使用带参数的cache操做能够设定RDD的缓存优先级。

6.3 rddbg：RDD程序的调试工具

RDD的初衷是为了实现容错以可以再计算（re-computation），这个特性使得调试更容易。咱们建立了一个名为rddbg的调试工具，它是经过基于程序记录的Lineage信息来实现的，容许用户：（1）重建任何由程序建立的RDD，并执行交互式查询；（2）使用一个单进程Java调试器（如jdb）传入计算好的RDD分区，可以从新运行做业中的任何任务。

咱们强调一下，rddbg不是一个彻底重放的调试器：特别是不对非肯定性的代码或动做进行重放。但若是某个任务一直运行很慢（好比因为数据分布不均匀或者异常输入等缘由），仍然能够用它来帮助找到其中的逻辑错误和性能问题。

举个例子，咱们使用rddbg去解决用户Spam分类做业中的一个bug，这个做业中的每次迭代都产生0值。在调试器中从新执行reduce任务，很快就能发现，输入的权重向量（存储在一个用户自定义的向量类中）居然是空值。因为从一个未初始化的稀疏向量中读取老是返回0，运行时也不会抛出异常。在这个向量类中设置一个断点，而后运行这个任务，引导程序很快就运行到设置的断点处，咱们发现向量类的一个数组字段的值为空，咱们诊断出了这个bug：稀疏向量类中的数据字段被错误地使用transient来修饰，致使序列化时忽略了该字段的数据。

rddbg给程序执行带来的开销很小。程序原本就须要将各个RDD中的全部闭包序列化并经过网络传送，只不过使用rddbg同时还要将这些闭集记录到磁盘。

7. 评估

咱们在Amazon EC2[1]上进行了一系列实验来评估Spark及RDD的性能，并与Hadoop及其余应用程序的基准进行了对比。总的说来，结果以下：
（1）对于迭代式机器学习应用，Spark比Hadoop快20多倍。这种加速比是由于：数据存储在内存中，同时Java对象缓存避免了反序列化操做。
（2）用户编写的应用程序执行结果很好。例如，Spark分析报表比Hadoop快40多倍。
（3）若是节点发生失效，经过重建那些丢失的RDD分区，Spark可以实现快速恢复。
（4）Spark可以在5-7s延时范围内，交互式地查询1TB大小的数据集。
咱们基准测试首先从一个运行在Hadoop上的具备迭代特征的机器学习应用（7.1）和PageRank（7.2）开始，而后评估在Spark中当工做集不能适应缓存（7.4）时系统容错恢复能力（7.3），最后讨论用户应用程序（7.5）和交互式数据挖掘（7.6）的结果。
除非特殊说明，咱们的实验使用m1.xlarge EC2 节点，4核15GB内存，使用HDFS做为持久存储，块大小为256M。在每一个做业运行执行时，为了保证磁盘读时间更加精确，咱们清理了集群中每一个节点的操做系统缓存。

7.1 可迭代的机器学习应用

咱们实现了2个迭代式机器学习（ML）应用，Logistic回归和K-means算法，与以下系统进行性能对比：

• Hadoop：Hadoop 0.20.0稳定版。
• HadoopBinMem：在首轮迭代中执行预处理，经过将输入数据转换成为开销较低的二进制格式来减小后续迭代过程当中文本解析的开销，在HDFS中加载到内存。
• Spark：基于RDD的系统，在首轮迭代中缓存Java对象以减小后续迭代过程当中解析、反序列化的开销。

咱们使用同一数据集在相同条件下运行Logistic回归和K-means算法：使用400个任务（每一个任务处理的输入数据块大小为256M），在25-100台机器，执行10次迭代处理100G输入数据集（表4）。两个做业的关键区别在于每轮迭代单个字节的计算量不一样。K-means的迭代时间取决于更新聚类坐标耗时，Logistic回归是非计算密集型的，可是在序列化和解析过程当中很是耗时。
因为典型的机器学习算法须要数10轮迭代，而后再合并，咱们分别统计了首轮迭代和后续迭代计算的耗时，并从中发现，在内存中缓存RDD极大地加快了后续迭代的速度。

表4 用于Spark基准程序的数据

应用	数据描述	大小
Logistic回归	10亿9维点数据	100G
K-means	10亿10维点数据（k=10）	100G
PageRank	400万Wikipedia文章超连接图	49G
交互式数据挖掘	Wikipedia浏览日志（2008-10~2009-4）	1TB

首轮迭代。在首轮迭代过程当中，三个系统都是从HDFS中读取文本数据做为输入。图9中“First Iteration”显示了首轮迭代的柱状图，实验中Spark快于Hadoop，主要是由于Hadoop中的各个分布式组件基于心跳协议来发送信号带来了开销。HadoopBinMem是最慢的，由于它经过一个额外的MapReduce做业将数据转换成二进制格式。

图8 首轮迭代后Hadoop、HadoopBinMen、Spark运行时间对比

后续迭代。图9显示了后续迭代的平均耗时，图8对比了不一样聚类大小条件下耗时状况，咱们发如今100个节点上运行Logistic回归程序，Spark比Hadoop、HadoopBinMem分别快25.三、20.7倍。从图8（b）能够看到，Spark仅仅比Hadoop、HadoopBinMem分别快1.九、3.2倍，这是由于K-means程序的开销取决于计算（用更多的节点有助于提升计算速度的倍数）。

后续迭代中，Hadoop仍然从HDFS读取文本数据做为输入，因此从首轮迭代开始Hadoop的迭代时间并无明显的改善。使用预先转换的SequenceFile文件（Hadoop内建的二进制文件格式），HadoopBinMem在后续迭代中节省了解析的代价，可是仍然带来的其余的开销，如从HDFS读SequenceFile文件并转换成Java对象。由于Spark直接读取缓存于RDD中的Java对象，随着聚类尺寸的线性增加，迭代时间大幅降低。

图9：首轮及其后续迭代平均时间对比
理解速度提高。咱们很是惊奇地发现，Spark甚至赛过了基于内存存储二进制数据的Hadoop（HadoopBinMem），幅度高达20倍之多，Hadoop运行慢是因为以下几个缘由：

1. Hadoop软件栈的最小开销
2. 读数据时HDFS栈的开销
3. 将二进制记录转换成内存Java对象的代价

为了估测1，咱们运行空的Hadoop做业，仅仅执行做业的初始化、启动任务、清理工做就至少耗时25秒。对于2，咱们发现为了服务每个HDFS数据块，HDFS进行了屡次复制以及计算校验和操做。

为了估测3，咱们在单个节点上运行了微基准程序，在输入的256M数据上计算Logistic回归，结果如表5所示。首先，在内存中的HDFS文件和本地文件的不一样致使经过HDFS接口读取耗时2秒，甚至数据就在本地内存中。其次，文本和二进制格式输入的不一样形成了解析耗时7秒的开销。最后，预解析的二进制文件转换为内存中的Java对象，耗时3秒。每一个节点处理多个块时这些开销都会累积起来，然而经过缓存RDD做为内存中的Java对象，Spark只须要耗时3秒。

表5 Logistic回归迭代时间

	内存中的HDFS文件	内存中的本地文件	缓存的RDD
文本输入 二进制输入	15.38 (0.26) 8.38 (0.10)	13.13 (0.26) 6.86 (0.02)	2.93 (0.31) 2.93 (0.31)

7.2 PageRank

经过使用存储在HDFS上的49G Wikipedia导出数据，咱们比较了使用RDD实现的Pregel与使用Hadoop计算PageRank的性能。PageRank算法经过10轮迭代处理了大约400万文章的连接图数据，图10显示了在30个节点上，Spark处理速度是Hadoop的2倍多，改进后对输入进行Hash分区速度提高到2.6倍，使用Combiner后提高到3.6倍，这些结果数据也随着节点扩展到60个时同步放大。

图10 迭代时间对比

7.3 容错恢复

基于K-means算法应用程序，咱们评估了在单点故障（SPOF）时使用Lneage信息建立RDD分区的开销。图11显示了，K-means应用程序运行在75个节点的集群中进行了10轮迭代，咱们在正常操做和进行第6轮迭代开始时一个节点发生故障的状况下对耗时进行了对比。没有任何失败，每轮迭代启动了400个任务处理100G数据。

图11 SPOF时K-means应用程序迭代时间
第5轮迭代结束时大约耗时58秒，第6轮迭代时Kill掉一个节点，该节点上的任务都被终止（包括缓存的分区数据）。Spark调度器调度这些任务在其余节点上从新并行运行，而且从新读取基于Lineage信息重建的RDD输入数据并进行缓存，这使得迭代计算耗时增长到80秒。一旦丢失的RDD分区被重建，平均迭代时间又回落到58秒。

7.4 内存不足时表现

到如今为止，咱们能保证集群中的每一个节点都有足够的内存去缓存迭代过程当中使用的RDD，若是没有足够的内存来缓存一个做业的工做集，Spark又是如何运行的呢？在实验中，咱们经过在每一个节点上限制缓存RDD所须要的内存资源来配置Spark，在不一样的缓存配置条件下执行Logistic回归，结果如图12。咱们能够看出，随着缓存的减少，性能平缓地降低。

图12 Spark上运行Logistic回归的性能表现

7.5 基于Spark构建的用户应用程序

In-Memory分析。视频分发公司Conviva使用Spark极大地提高了为客户处理分析报告的速度，之前基于Hadoop使用大约20个Hive[3]查询来完成，这些查询做用在相同的数据子集上（知足用户提供的条件），可是在不一样分组的字段上执行聚合操做（SUM、AVG、COUNT DISTINCT等）须要使用单独的MapReduce做业。该公司使用Spark只须要将相关数据加载到内存中一次，而后运行上述聚合操做，在Hadoop集群上处理200G压缩数据并生成报耗时20小时，而使用Spark基于96G内存的2个节点耗时30分钟便可完成，速度提高40倍，主要是由于不须要再对每一个做业重复地执行解压缩和过滤操做。

城市交通建模。在Berkeley的Mobile Millennium项目[17]中，基于一系列分散的汽车GPS监测数据，研究人员使用并行化机器学习算法来推算公路交通拥堵情况。数据来自市区10000个互联的公路线路网，还有600000个由汽车GPS装置采集到的样本数据，这些数据记录了汽车在两个地点之间行驶的时间（每一条路线的行驶时间可能跨多个公路线路网）。使用一个交通模型，经过推算跨多个公路网行驶耗时预期，系统可以估算拥堵情况。研究人员使用Spark实现了一个可迭代的EM算法，其中包括向Worker节点广播路线网络信息，在E和M阶段之间执行reduceByKey操做，应用从20个节点扩展到80个节点（每一个节点4核），如图13（a）所示：

图13 每轮迭代运行时间（a）交通建模应用程序（b）基于Spark的社交网络的Spam分类
社交网络Spam分类。Berkeley的Monarch项目[31]使用Spark识别Twitter消息上的Spam连接。他们在Spark上实现了一个相似7.1小节中示例的Logistic回归分类器，不一样的是使用分布式的reduceByKey操做并行对梯度向量求和。图13（b）显示了基于50G数据子集训练训练分类器的结果，整个数据集是250000的URL、至少10^7个与网络相关的特征/维度，内容、词性与访问一个URL的页面相关。随着节点的增长，这并不像交通应用程序那样近似线性，主要是由于每轮迭代的固定通讯代价较高。

7.6 交互式数据挖掘

为了展现Spark交互式处理大数据集的能力，咱们在100个m2.4xlarge EC2实例（8核68G内存）上使用Spark分析1TB从2008-10到2009-4这段时间的Wikipedia页面浏览日志数据，在整个输入数据集上简单地查询以下内容以获取页面浏览总数：（1）所有页面；（2）页面的标题能精确匹配给定的关键词；（3）页面的标题能部分匹配给定的关键词。

图14 显示了分别在整个、1/二、1/10的数据上查询的响应时间，甚至1TB数据在Spark上查询仅耗时5-7秒，这比直接操做磁盘数据快几个数量级。例如，从磁盘上查询1TB数据耗时170秒，这代表了RDD缓存使得Spark成为一个交互式数据挖掘的强大工具。

8. 相关工做

分布式共享内存（DSM）。RDD能够当作是一个基于DSM研究[24]获得的抽象。在2.5节咱们讨论过，RDD提供了一个比DSM限制更严格的编程模型，并能在节点失效时高效地重建数据集。DSM经过检查点[19]实现容错，而Spark使用Lineage重建RDD分区，这些分区能够在不一样的节点上从新并行处理，而不须要将整个程序回退到检查点再从新运行。RDD可以像MapReduce同样将计算推向数据[12]，并经过推测执行来解决某些任务计算进度落后的问题，推测执行在通常的DSM系统上是很难实现的。

In-Memory集群计算。Piccolo[28]是一个基于可变的、In-Memory的分布式表的集群编程模型。由于Piccolo容许读写表中的记录，它具备与DSM相似的恢复机制，须要检查点和回滚，可是不能推测执行，也没有提供相似groupBy、sort等更高级别的数据流算子，用户只能直接读取表单元数据来实现。可见，Piccolo是比Spark更低级别的编程模型，可是比DSM要高级。

RAMClouds[26]适合做为Web应用的存储系统，它一样提供了细粒度读写操做，因此须要经过记录日志来实现容错。

数据流系统。RDD借鉴了DryadLINQ[34]、Pig[25]和FlumeJava[9]的“并行收集”编程模型，经过容许用户显式地将未序列化的对象保存在内存中，以此来控制分区和基于key随机查找，从而有效地支持基于工做集的应用。RDD保留了那些数据流系统更高级别的编程特性，这对那些开发人员来讲也比较熟悉，并且，RDD也可以支持更多类型的应用。RDD新增的扩展，从概念上看很简单，其中Spark是第一个使用了这些特性的系统，相似DryadLINQ编程模型，可以有效地支持基于工做集的应用。

面向基于工做集的应用，已经开发了一些专用系统，像Twister[13]、HaLoop[8]实现了一个支持迭代的MapReduce模型；Pregel[21]，支持图应用的BSP计算模型。RDD是一个更通用的抽象，它可以描述支持迭代的MapReduce、Pregel，还有现有一些系统未能处理的应用，如交互式数据挖掘。特别地，它可以让开发人员动态地选择操做来运行在RDD上（如查看查询的结果以决定下一步运行哪一个查询），而不是提供一系列固定的步骤去执行迭代，RDD还支持更多类型的转换。

最后，Dremel[22]是一个低延迟查询引擎，它面向基于磁盘存储的大数据集，这类数据集是把嵌套记录数据生成基于列的格式。这种格式的数据也可以保存为RDD并在Spark系统中使用，但Spark也具有将数据加载到内存来实现快速查询的能力。

Lineage。咱们经过参考[6]到[10]作过调研，在科学计算和数据库领域，对于一些应用，如须要解释结果以及容许被从新生成、工做流中发现了bug或者数据集丢失须要从新处理数据，表示数据的Lineage和原始信息一直以来都是一个研究课题。RDD提供了一个受限的编程模型，在这个模型中使用细粒度的Lineage来表示是很是容易的，所以它能够被用于容错。

缓存系统。Nectar[14]可以经过识别带有程序分析的子表达式，跨DryadLINQ做业重用中间结果，若是将这种能力加入到基于RDD的系统会很是有趣。可是Nectar并无提供In-Memory缓存，也不可以让用户显式地控制应该缓存那个数据集，以及如何对其进行分区。Ciel[23]一样可以记住任务结果，但不能提供In-Memory缓存并显式控制它。

语言迭代。DryadLINQ[34]可以使用LINQ获取到表达式树而后在集群上运行，Spark系统的语言集成与它很相似。不像DryadLINQ，Spark容许用户显式地跨查询将RDD存储到内存中，并经过控制分区来优化通讯。Spark支持交互式处理，但DryadLINQ却不支持。

关系数据库。从概念上看，RDD相似于数据库中的视图，缓存RDD相似于物化视图[29]。然而，数据库像DSM系统同样，容许典型地读写全部记录，经过记录操做和数据的日志来实现容错，还须要花费额外的开销来维护一致性。RDD编程模型经过增长更多限制来避免这些开销。

9. 总结

咱们提出的RDD是一个面向，运行在普通商用机集群之上并行数据处理应用的分布式内存抽象。RDD普遍支持基于工做集的应用，包括迭代式机器学习和图算法，还有交互式数据挖掘，然而它保留了数据流模型中引人注目的特色，如自动容错恢复，处理执行进度落后的任务，以及感知调度。它是经过限制编程模型，进而容许高效地重建RDD分区来实现的。RDD实现处理迭代式做业的速度超过Hadoop大约20倍，并且还可以交互式查询数百G数据。

致谢

首先感谢Spark用户，包括Timothy Hunter、Lester Mackey、Dilip Joseph、Jibin Zhan和Teodor Moldovan，他们在真实的应用中使用Spark，提出了宝贵的建议，同时也发现了一些新的研究挑战。此次研究离不开如下组织或团体的大力支持：Berkeley AMP Lab创立赞助者Google和SAP，AMP Lab赞助者Amazon Web Services、Cloudera、Huawei、IBM、Intel、Microsoft、NEC、NetApp和VMWare，国家配套资金加州MICRO项目（助学金 06-152，07-010），国家天然科学基金 （批准 CNS-0509559），加州大学工业/大学合做研究项目 （UC Discovery）授予的COM07-10240，以及天然科学和加拿大工程研究理事会。

转自：http://shiyanjun.cn/archives/744.html