你好我是程序员贵哥。过去几讲里无论是Hive这样基于MapReduce的系统还是Dremel这样抛开MapReduce的系统其实都已经反映了MapReduce这个大数据处理的计算模型在2010年这个时间节点已经有一些“落后”了。来自Facebook的Hive选择了在MapReduce上优化改良仍然基于MapReduce的模型。而Google自家的Dremel则是另起炉灶用一个新的底层架构来支持OLAP的数据分析。不过在工业界之外学术界一样不会在整个大数据飞速发展的时代里缺席。2010年来自Berkeley的博士生Matei Zaharia发表了一篇论文《Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing》。伴随着这篇论文的是一个开源系统也就是Spark。在之后的几年里Spark不仅逐步侵蚀了Hadoop的市场份额成为大数据分析的第一选择也在实时数据处理以及大规模机器学习里占据了重要的地位。Spark的计算模型其实可以看作一个是支持多轮迭代的MapReduce模型。不过它在实现层面又和稍显笨拙的MapReduce完全不同。通过引入RDD这样一个函数式对象的数据集的概念Spark在多轮的数据迭代里不需要像MapReduce一样反反复复地读写硬盘大大提升了处理数据的性能。最重要的是Spark的这篇引入了RDD概念的论文让我们看到了一个优秀的系统设计并不需要有多复杂Spark论文发表的时候整个系统也就1万多行代码一个人花上一两个月就能写出来。整个系统也没有太多复杂难懂的学术概念只是针对MapReduce明显的不足之处做了改造而已。那么接下来我们就一起来看看Spark是如何做到这一点的它的系统设计又有什么精妙之处。通过这一讲的学习我希望你能够掌握以下几点Spark里核心的RDD是一个什么概念它是通过什么方式来优化分布式数据处理的它的设计思路和计算机科学中的哪些观念比较相像。在系统设计层面我们怎么利用人的经验预先设计好系统针对正常情况下和异常情况下的性能进行权衡和选择的。使用硬盘来“容错”的MapReduce在深入到Spark的论文之前我们先来简单回顾一下MapReduce的整个流程。我把我们在讲解MapReduce的论文中用到的数据流程图放到了这里你可以再复习下。MapReduce的过程并不复杂Map函数的输出结果会输出到所在节点的本地硬盘上。Reduce函数会从Map函数所在的节点里拉取它所需要的数据然后再写入本地。接着通过一个外部排序的过程把数据进行分组。最后将排序分完组的数据通过Reduce函数进行处理。你会发现**在这个过程里任何一个中间环节我们都需要去读写硬盘。**Map函数的处理结果并不会直接通过网络发送给Reduce所在的Worker节点Reduce也不会直接在内存中做数据排序。这还只是单个的MapReduce任务如果我们要通过MapReduce来跑一些机器学习任务比如通过L-BFGS这样的算法来进行大规模的逻辑回归的模型训练我们需要跑上百个MapReduce的任务。其中每一个MapReduce的任务都差不多都是把所有日志读入然后和一个梯度向量做计算算出新的梯度。接着下一轮计算要再读入一遍原来的日志数据再和新的梯度做计算。在这个过程中我们原始的日志会被重复读取上百遍。而我们知道内存的读写速度是远远快于硬盘的这里我放了《深入浅出计算机组成原理》专栏里的一张数据表。可以看到无论你在用机械硬盘还是SSD硬盘访问数据的延时和内存比都有1000倍以上的差距。之所以MapReduce把所有数据都往硬盘里一写是因为它追求的是设计上的“简单”以及在大规模集群下的“容错”能力。把Map节点的输出数据直接通过网络发送给Reduce的确是一个很直观的想法让Reduce节点在内存里就直接处理数据也的确可以提升性能。但是如果在这个过程中Map或者Reduce的节点出现故障了怎么办因为Reduce对于前面的Map函数有依赖关系所以任何一个Map节点故障意味着Reduce只收到了部分数据而且它还不知道是哪一部分。那么Reduce任务只能失败掉然后等Map节点重新来过。而且Reduce的失败还会导致其他的Map节点计算的数据也要重来一遍引起连锁反应最终等于是整个任务重来一遍。这只是我们尝试让数据不需要落地到硬盘处理中会遇到的一种情况我们还可能遇到网络拥塞、内存不足以处理传输的数据等种种情况。事实上你可以认为传统的MPI分布式计算系统就是这样让一个节点直接往另外一个节点发送消息来传递数据的但是这样的系统容错能力很差所以集群的规模往往也上不去。而MapReduce针对这个问题的解决方案非常简单粗暴那就是把整个数据处理的环节完全拆分开来然后把一个个阶段的中间数据都落地到硬盘上。这样针对单个节点的故障我们只需要重新运行对应节点分配的任务就好了其他已经完成的工作不会“半途而废”。分布式系统里的“函数式”数据集但是这样的方式真的是效率最高的吗虽然因为整个GFS和MapReduce系统规模很大节点很多我们的各个节点必然会出现故障。但是毕竟硬件也好、系统也好出现故障是小概率事件。我们在第8讲里看过一个数据中心里的18万块硬盘90天坏了439块算到年度的损坏率也就是1%左右。我们为了这1%的错误却要把99%的数据都反复从硬盘里读出来写进去感觉的确有点划不来。所以很自然地我们需要有一个更有效率的容错方式。一个很直观的想法自然也就冒出来了那就是我们是否可以做这三件事情第一个是我们是否可以把数据缓存在内存里。如果是计算的中间结果我们不一定要把它写到硬盘上。如果是反复读取的输入数据我们可以缓存在内存里而不是每个迭代重新读取一遍。第二个是可以记录我们运算数据生成的“拓扑图”。也就是记录数据计算的依赖关系一旦某个节点故障导致这个依赖关系中的一部分节点出现故障我们根据拓扑图重新计算这一部分数据就好了。通过这样的方式来解决容错问题而不是每一次都把数据写入到硬盘。第三个是可以通过检查点来在特定环节把数据写入到硬盘。当我们的拓扑图层数很深或者数据要反复进行很多次的迭代计算。前面通过“拓扑图”进行重新计算的容错方式会变得非常低效那么我们可以在一部分中间环节把数据写入到硬盘上。这样一种折衷的方式既避免了每次都去从硬盘读写数据也避免了一旦某一个环节出现故障“容错”方案只能完全从头再来的尴尬出现。举一个具体的例子如果前面提到的大规模逻辑回归的机器模型训练需要进行100轮迭代。我们最好的解决方案既不是每轮迭代都需要重新读写数据那样太浪费硬盘的I/O了。也不是把几小时的计算过程都放在内存里那样万一计算梯度结果的数据在第99轮丢失了我们就要从头开始。我们完全可以把读取的日志数据缓存在内存里然后把每10轮计算完的梯度数据写入到硬盘上。这样一旦出现故障我们只需要重新读取一次日志数据并最多计算10轮迭代的过程就好了。valpointsspark.textFile(...).map(parsePoint).persist()varw// random initial vectorfor(i-1to ITERATIONS){valgradientpoints.map{pp.x*(1/(1exp(-p.y*(w dot p.x)))-1)*p.y}.reduce((a,b)ab)w-gradient}论文3.2.1 部分使用Spark实现分布式逻辑回归的代码。我们的输入数据会通过.persistent缓存在内存中而不需要每个迭代都重新从硬盘读取。其实这个思路就是Spark整个系统设计的出发点。根据这个朴素的思路Spark定了一个新的概念RDD全称是Resilient Distributed Dataset中文叫做弹性分布式数据集。整个系统设计的其实就是“弹性”“分布式”“数据集”这三点的组合。分布式和数据集这两个关键字我们应该不需要再进行太多的解释了。我们在GFS、Bigtable乃至Dremel里都采用了数据分区的方式来确保数据是分布式的。“函数式”的RDD而RDD最核心的设计关键点就在这个弹性上。论文里的2.1部分给出了RDD明确的定义这句话是这样说的RDD是只读的、已分区的记录集合RDD只能通过明确的操作以及通过两种数据创建稳定存储系统中的数据其他RDD。这个明确的操作是指map、filter和join这样的操作以和其他的操作区分开来。按照这个定义我们可以看到这个是对于数据的一个抽象。我们的任何一个数据集进行一次转换就是一个新的RDD但是这个RDD并不需要实际输出到硬盘上。实际上这个数据都不会作为一个完整的数据集缓存在内存中而只是一个RDD的“抽象概念”。只有当我们对某一个RDD实际调用persistent函数的时候这个RDD才会实际作为一个完整的数据集缓存在内存中。一旦被缓存到内存里这个RDD就能够再次被下游的其他数据转换反复使用。一方面这个数据不需要写入到硬盘所以我们减少了一次数据写。另一方面下游的其他转化也不需要再从硬盘读数据于是我们就节省了大量的硬盘I/O的开销。linesspark.textFile(hdfs://...)errorslines.filter(_.startsWith(ERROR))errors.persist()// 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()我们可以对照着论文2.2.1中的示例代码来看这样一个过程。我们从HDFS上读入原始数据根据关键词ERROR进行了一次过滤然后把它persistent下来。而接下来分别有两个分析任务会用到这个缓存在内存里的ERROR数据一个是找出所有带有MySQL关键词的错误日志然后进行统计行数另一个则是找到所有带有HDFS关键字的日志然后按照Tab分割并收集第3列的数据。这里面一开始从HDFS里读入的line数据因为没有persistent所以不会缓存在内存中。而errors会缓存在内存里面供后面两个任务作为输入使用。errors我们不需要写入到硬盘里而后面分析MYSQL和HDFS关键字错误的两个任务也不需要从硬盘读数据数据都是直接在内存中读写所以性能大大加快了。从RDD的这个逻辑上其实我们可以看到计算机工程上的其他系统中的影子。第一个是惰性求值Lazy-Evaluation我们的一层层数据转化只要没有调用persistent都可以先不做计算而只是记录这个计算过程中的函数。而当persistent一旦被调用那么我们就需要把实际的数据结果计算出来并存储到内存里再供后面的数据转换程序调用。第二个是数据库里的视图功能。为了查询方便对于复杂的多表关联很多时候我们会预先建好一张数据库的逻辑视图。那么我们在查询逻辑视图的时候其实还是通过一个多表关联SQL去查询原始表的这个就好像我们并没有调用persistent把数据实际持久化下来。当然我们也可以把对应视图的查询结果直接写入一张中间表这样实际上就相当于把计算的结果持久化下来了后续查询的SQL就会查询这个中间表。如果视图里的数据会被后续的SQL反复多次查询并且对应的原始数据集也和RDD一样是不可变的话一样会大大提升系统整体的效率。宽依赖关系和检查点那么通过调用persistent来把数据缓存到内存里就减少了大量的硬盘读写。但是我们仍然会面临节点失效导致RDD需要重新计算的情况。所以Spark对这部分流程做了进一步的优化这个优化说起来其实也不复杂。那就是如果一个节点失效了导致的数据重新计算需要影响的节点太多那么我们就把计算结果输出到硬盘上。而如果影响的节点少那么我们就只是单独重新计算被响应到的那些节点就好了。所以在Spark里会对整个数据计算的拓扑图在分布式系统下的依赖关系做一个分类。如果一个RDD的一个分区只会影响到下游的一个节点那么我们就称这样的上下游依赖关系为窄依赖。而如果一个RDD的一个分区会影响到下游的多个节点那么我们就称这样的上下游关系为宽依赖。对于窄依赖即使重算一遍也只是影响一条线上的少数几个节点所以对应的中间数据结果并不需要输出到硬盘上。而对于宽依赖一旦上游的一个节点失效了需要重新计算。那么它对应的多个下游节点都需要重新从这个节点拉取数据并重新计算需要占用更多的网络带宽和计算资源。换句话说在宽依赖下一个上游节点的失效会以几倍的影响在下游得到放大。所以在宽依赖的场景下上游会像MapReduce里的Map一样把输出结果序列化到硬盘上以减少故障后的恢复成本。同样的对于有多轮迭代或者是整个拓扑图很长的数据处理任务Spark在persistent的时候支持你添加一个REPLICATE参数把当前的计算结果作为一个检查点存储下来。一旦添加了这个参数数据就不只是存储在内存中而是会序列化到硬盘里。这样同样可以减少你在出现故障时候的重新计算的时间。可以看到无论是persistent、宽依赖下的数据会被持久化存储还是允许用户去自己通过检查点存储中间步骤的计算结果都是为了在日常情况下的性能和容错情况下的性能做一个平衡。如果不做任何持久化存储那么平时系统会跑得很快但是一旦某个节点出错就要从头再来。而如果都做持久化存储那么节点出错的时候计算可以恢复得很快但是没有问题的时候会有很多浪费。本质上Spark就是根据人的经验在这两点上做好了平衡。小结其实Spark里的RDD的设计思想并不复杂。和MapReduce一样RDD的设计思路也是来自于函数式编程。相对于过程式地把每一个数据转换Transformation的结果存储下来RDD相当于记录了输入数据以及对应的计算输入数据的函数。这个方式和把一步步的计算结果存储下来的效果一样都可以解决容错问题。当某一个RDD的某一个分区因为故障丢失的时候我们可以通过输入数据和对应的函数快速计算出当前RDD的实际内容。而这个输入数据对应函数的组合就是RDD中的Lineage图。RDD和其他分布式系统最大的差异就在代表弹性的R这个关键字上。这个弹性体现在两个方面。**第一个是数据存储上。**数据不再是存放在硬盘上的而是可以缓存在内存中。只有当内存不足的时候才会把它们换出到硬盘上。同时数据的持久化也支持硬盘、序列化后的内存存储以及反序列化后Java对象的内存存储三种形式。每一种都比另一种需要占用更多的内存但是计算速度会更快。**第二个是选择把什么数据输出到硬盘上。**Spark会根据数据计算的Lineage来判断某一个RDD对于前置数据是宽依赖还是窄依赖的。如果是宽依赖的意味着一个节点的故障可能会导致大量的数据要进行重新计算乃至数据网络传输的需求。那么它就会把数据计算的中间结果存储到硬盘上。同时Spark也支持你自己定义检查点你可以把一些关键节点的数据通过检查点的方式持久化到硬盘上避免出现特定节点的故障导致大量数据需要重新计算的问题。可以看到Spark的RDD是在没有破坏MapReduce的易用性的前提下支持了MapReduce可以支持的所有运算方式。并且它通过尽可能利用内存使得需要多个MapReduce的组合或者迭代的任务的执行速度大大加快了。从分布式的逻辑回归来看Spark的性能会比使用原始的MapReduce一轮轮迭代快上20倍。这也是为什么Spark一出现在市场上就很快替代了大量的MapReduce的分析工作并在迭代式的机器学习算法中成为了主流。