Fighting bird分享 http://blog.sciencenet.cn/u/tonia

博文

弹性分布式数据集:基于内存的集群计算的容错性抽象(1)

已有 11562 次阅读 2011-12-22 14:09 |个人分类:译文|系统分类:科研笔记|关键词:spark| spark

弹性分布式数据集:基于内存的集群计算的容错性抽象(1)—— 引言、RDD

弹性分布式数据集:基于内存的集群计算的容错性抽象(2)—— 编程接口、RDD实现实例、任务调度

弹性分布式数据集:基于内存的集群计算的容错性抽象(3)—— 实现、实验结果


摘要:本文提出了弹性分布式数据集(RDD,Resilient Distributed Datasets),这是一种分布式的内存抽象,允许在大型集群上执行基于内存的计算(In-Memory Computing),与此同时还保持了MapReduce等数据流模型的容错特性。现有的数据流系统对两种应用的处理并不高效:一是迭代式算法,这在图应用和机器学习领域很常见;二是交互式数据挖掘工具。这两种情况下,将数据保存在内存中能够极大地提高性能。为了有效地实现容错,RDD提供了一种高度受限的共享内存,即RDD是只读的,并且只能通过其他RDD上的批量操作来创建。尽管如此,RDD仍然足以表示很多类型的计算,包括MapReduce和专用的迭代编程模型(如Pregel)等。我们实现RDD在迭代计算方面比Hadoop快二十多倍,同时还可以在5-7秒的延时内交互式地查询1TB的数据集。


1.引言


无论是工业界还是学术界,都已经广泛使用高级集群编程模型来处理日益增长的数据,如MapReduceDryad。这些系统将分布式编程简化为自动提供位置感知性(locality-aware)调度、容错以及负载均衡,使得大量用户能够在商用集群上分析庞大的数据集。


大多数现有的集群计算系统都是基于非循环的数据流模型(acyclic data flow model)。从稳定的物理存储(如分布式文件系统)中加载记录,一组确定性操作构成一个DAG,记录被传入这个DAG,然后写回稳定存储。通过这个DAG数据流图,运行时自动完成调度工作及故障恢复。


尽管非循环数据流是一种很强大的抽象方法,但仍然有些应用无法使用这种方式描述。我们就是针对这些不太适合非循环模型的应用,它们的特点是在多个并行操作之间重用工作数据集(working set)。这类应用包括:(1)机器学习和图应用中常用的迭代算法(每一步对数据执行相似的函数);(2)交互式数据挖掘工具(用户反复查询一个数据子集)。基于数据流的架构并不明确支持工作集,所以需要将数据输出到磁盘然后在每次查询时重新加载,从而带来较大的开销。


我们提出了一种分布式的内存抽象,称为弹性分布式数据集(RDDResilient Distributed Datasets),支持基于工作集的应用,同时具有数据流模型的特点:即自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显式地将工作集缓存在内存中,极大地加速了后期的工作集重用。


RDD提供了一种高度受限的共享内存方式,即RDD是只读的记录分区的集合,只能通过对其他RDD执行确定性的转换操作(如mapjoingroup by)而创建。这些限制保证了低开销的容错性。与分布式共享内存系统需要高成本的检查点(checkpoint)和回滚(rollback)不同,RDD通过血统(lineage)来重建丢失的分区:RDD中包含如何从其他RDD衍生(即计算)出本RDD所需的相关信息,这样不需要检查点操作就可以重新构建丢失的数据分区。尽管RDD不是一个普适的共享内存抽象,但却具备了良好的描述能力、可伸缩性和可靠性,非常适合大多数数据并行型应用。


第一个指出非循环数据流存在不足的并不是我们。例如,GooglePregel,是一种专门用于迭代式图算法的编程模型;TwisterHaLoop,是两种典型的迭代式MapReduce模型。但是,这些系统都是针对特定类型应用的。相比之下,RDD则为基于工作集的应用提供了更为通用的抽象。用户可以对中间结果进行显式的命名和物化(materialize),控制其分区,然后使用它们执行特定的用户操作(而不是让运行时去循环执行一系列MapReduce步骤)。RDD可以用来描述Pregel、迭代式MapReduce,以及这两种模型无法描述的其他应用,如交互式数据挖掘工具(用户将数据集装入RAM,然后执行ad-hoc查询)。


我们实现RDD称为Spark,用于开发多种并行应用。Spark采用Scala语言实现,提供类似于DryadLINQ的集成语言编程接口,方便用户编写并行任务。此外,通过修改Scala解释器,Spark还可用于交互式查询大数据集。我们相信Spark是第一个允许集群上对大数据集进行交互式分析的有效、通用的编程语言框架。


我们通过微基准(microbenchmark)和用户应用程序来评估RDD。实验表明,在处理迭代式应用上SparkHadoop快高达20多倍,数据分析报表的性能提高了40多倍,同时能够在5-7秒的延时内交互式扫描1TB的数据集。此外,我们还在Spark之上实现了PregelHaLoop编程模型(包括其位置优化策略,placement optimization),以库的形式实现(分别使用了100200Scala代码)。最后,利用RDD内在的确定性特性,我们还创建了一种Spark调试工具rddbg,允许用户在任务期间利用血统关系(lineage)重建RDD,然后像传统调试器那样重新执行任务。


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

 

2.弹性分布式数据集(RDD

本部分描述RDD和编程模型。首先讨论设计目标(2.1),然后定义RDD2.2),接着讨论Spark的编程模型(2.3),并给出一个示例(2.4),最后将RDD与分布式共享内存进行比较(2.5)。


2.1 目标和概述

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


在这些特性中,最难实现的是容错性。一般来说,分布式数据集的容错性有两种方式:即数据检查点和记录数据的更新。我们面向的是大规模数据分析,数据检查点操作成本很高:需要通过数据中心的网络连接在机器之间复制庞大的数据集,而网络带宽往往比内存带宽低得多,同时还需要消耗更多的存储资源(在RAM中复制数据可以减少需要缓存的数据量,而存储到磁盘则会拖慢应用程序)。所以,我们选择记录更新的方式。但是,如果更新太多,那么记录更新成本也不低。因此,RDD只支持粗粒度转换(coarse-grained transformation),即在大量记录上执行的单个操作。将创建RDD的一系列转换记录下来(即lineage),以便恢复丢失的分区。


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


2.2 RDD抽象

RDD是只读的记录分区的集合。RDD只能通过在——(1)稳定物理存储中的数据集;(2)其他已有的RDD——上执行确定性(deterministic)操作来创建。这些操作称之为转换(transformation),如map, filter, groupBy, join。(转换不是程序员在RDD上执行的操作。)


RDD不需要物化。RDD含有如何从其他RDD衍生(即计算)出本RDD的相关信息(即lineage),据此可以从物理存储的数据计算出相应的RDD分区(partition)。


2.3 编程模型

Spark中,RDD被表示为对象,通过这些对象上的方法(或函数)调用转换(transformation)。


定义RDD之后,程序员就可以在行为(action)中使用RDD了。行为(action)是向应用程序返回值,或向存储系统导出数据的那些操作,例如,count(返回RDD中的元素个数),collect(返回元素本身),save(将RDD输出到存储系统)。在Spark中,只有在action第一次使用RDD时,才会计算RDD(即懒计算,lazily evaluated)。这样在构建RDD的时候,运行时以流水线的方式执行(pipeline)多个转换。


程序员还可以从两个方面控制RDD,即缓存(caching)和分区(partitioning)。用户可以请求将RDD缓存,这样运行时将已经计算好的RDD分区存储起来,以加速后期的重用。缓存的RDD一般存储在内存中,但如果内存不够,可以溢出(spill)到磁盘。


另一方面,RDD还允许用户根据关键字(key)指定分区顺序,这是一个可选的功能。目前支持哈希分区(hash partition)和范围分区(range partition)。例如,应用程序请求将两个RDD按照同样的哈希分区方式进行分区(将同一机器上具有相同关键字的记录放在一个分区),以加速它们之间的join操作。在PregelHaLoop中,多次迭代之间采用一致性的分区放置策略(Consistent partition placement)进行优化,我们同样也允许用户指定这种优化。


2.4 示例:控制台日志挖掘

本部分我们通过一个具体示例来阐述RDD。假定有一个大型网站出错,操作员想要检查Hadoop文件系统(HDFS)中的日志文件(TB级大小)来找出原因。通过使用Spark,操作员只需将日志中的错误信息装载到一组节点的RAM中,然后执行交互式查询。首先她需要在Spark解释器中敲入以下Scala命令:


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

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

errors.cache()


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


这时集群还没有开始执行任何任务。但是,用户已经可以在这个RDD上执行action操作了,例如统计错误消息的数目:

errors.count()

 

用户还可以在RDD上执行更多的转换(transformation)操作,并使用转换结果,如:

// 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()

 

使用errors的第一个action运行以后,Spark会把errors的分区缓存在内存中,极大地加速了后续计算。注意,最初的RDD lines不会被缓存。因为错误信息可能只占原数据集的很小一部分(小到足以放入内存)。


最后,为了说明模型的容错性,图1给出了第3个查询的血统(lineage)关系图。在lines RDD上执行filter操作,得到errors,然后再filtermap后得到新的RDD,在这个RDD上执行collect行为。Spark调度器以流水线的方式执行后两个转换,向拥有errors分区缓存的节点发送一组任务。此外,如果某个errors分区丢失,Spark只在相应的lines分区上执行filter操作来重建该errors分区。


 (由于粘贴不便,图参见论文原文

1 示例中第三个查询的血统关系图。(方框表示RDD,箭头表示转换)

 

2.5 RDD与分布式共享内存

为了进一步理解RDD是一种分布式的内存抽象,表1列出了RDD与分布式共享内存(DSMdistributed shared memory)的对比。在DSM系统中,应用可以向全局地址空间的任意位置进行读写操作。(注意这里的DSM,不仅指传统的共享内存系统,还包括那些通过分布式哈希表或分布式文件系统进行数据共享的系统,比如Piccolo。)DSM是一种通用的抽象,但这种通用性同时也使得在商用集群上实现有效的容错性更加困难。


RDDDSM主要区别在于,不仅可以通过批量转换创建(即“写”)RDD,还可以对任意内存位置读写。也就是说,RDD限制应用执行批量写操作,这样有利于实现有效的容错。特别地,RDD没有检查点开销,因为可以使用lineage来恢复RDD。而且,失效时只需要重新计算丢失的那些RDD分区,可以在不同节点上并行执行,而不需要回滚(roll back)整个程序。

 

对比项目

RDD

分布式共享内存

批量或细粒度操作

细粒度操作

批量转换操作

细粒度操作

一致性

不重要(RDD是不可更改的)

取决于应用程序或运行时

容错性

细粒度,低开销(使用lineage

需要检查点操作和程序回滚

落后任务的处理

任务备份

很难处理

任务安排

基于数据存放的位置自动实现

取决于应用程序(通过运行时实现透明性)

如果内存不够

与已有的数据流系统类似

性能较差(交换?)

1 RDD与分布式共享内存的对比

 

注意,通过备份任务的拷贝,RDD还可以处理落后任务(即运行很慢的节点),这点与MapReduce类似。而DSM则难以实现备份任务,因为任务及其副本都需要读写同一个内存位置。


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


最后看一下读操作的粒度。RDD上的很多行为(action,如countcollect)都是批量读操作,即扫描整个数据集,可以将任务分配到距离数据最近的节点上。同时,RDD也支持细粒度操作,即在哈希或范围分区的RDD上执行关键字查找。

(未完待续)


本系列文章《弹性分布式数据集:基于内存的集群计算的容错性抽象》系《Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing》译文。原文参见这里



http://blog.sciencenet.cn/blog-425672-520947.html

上一篇:IBM 技术文档:Spark, 快速数据分析的又一选择
下一篇:弹性分布式数据集:基于内存的集群计算的容错性抽象(2)

2 黄富强 crossludo

该博文允许注册用户评论 请点击登录 评论 (0 个评论)

数据加载中...

Archiver|手机版|科学网 ( 京ICP备14006957 )

GMT+8, 2019-5-21 18:40

Powered by ScienceNet.cn

Copyright © 2007- 中国科学报社

返回顶部