从WordCount看Spark大数据处理的核心机制

大微微---0170

从WordCount看Spark大数据处理的核心机制

　　大数据处理肯定是分布式的了，那就面临着几个核心问题：可扩展性，负载均衡，容错处理。Spark是如何处理这些问题的呢?接着上一篇的“动手写WordCount”，今天要做的就是透过这个大数据界的HelloWorld来看看Spark隐藏了哪些魔法。


　　请各位看官，带着分布式的问题往下看。
　　分布式架构
　　大数据时代，单机装下PB级的数据，然后在可接受的时间内处理完，不可能，所以一定是分布式的。


　　▶ 分布式存储
　　HDFS(Hadoop Distributed File System)是最常见的，和Spark配合的分布式存储系统。HDFS的存储结构如下图


　　每个文件被分成固定大小的块，而块作为最小的存储单位放到众多服务器上。那一旦存某个块的机器挂了，不是整个文件就洗白了吗?HDFS当然不会这么傻，文件的每个块都有备份，默认情况下一个块会存3份，分到不同的服务器。这样一来，除非某个块涉及的三台服务器全挂，否则不用担心。在合理分布3个块的情况下，三台服务器全挂的可能性比中500万还低。下面是/file.txt有三个文件块的情况。

　　NN是Name Node，存储文件块放在哪儿等元信息。DN是Data Node，用来存放具体的文件块。


　　▶ 分布式处理
　　有一类系统数据是分布式存储，但是处理却集中在一起。比如Mysql分库分表存数据，然后在某个服务器上，挨个获取所有库所有表的数据进行处理，这种系统的本质还是“数据分发到计算逻辑侧”，它的性能瓶颈就在于做数据处理的那台服务器。


　　而分布式处理的核心观念在于“把计算逻辑分发到数据侧”，有两大优点：
　　计算逻辑分发明显比数据分发节省网络带宽，而网络带宽是分布式系统中最宝贵的资源
　　计算逻辑在数据侧执行，消除了集中式处理中计算逻辑侧的性能瓶颈
　　Spark + HDFS的运行架构如下：

　　Driver是程序开始运行的地方，也是总控，它把计算逻辑(闭包的实例)发送到有数据块的Slave上去执行，结果再收回去汇总。


　　是不是看出来了?
　　数据更多了，加机器呗，机器多了磁盘多，磁盘多了存的多。
　　跑的慢了，加机器呗，机器多了磁盘多，并行加载起来，数据吐吞量大。机器多了，内存CPU也多，并行处理起来，数据吞吐量大。
　　提示： 分布式处理系统会把计算逻辑分发到数据侧，极大提高系统的水平扩展性。


　　WordCount运行机制
　　讲了一堆理论知识，为了让各位看官透彻理解，也为Spark程序算法优化打下坚实的基础，我们拿WordCount来举例说明，顺便说说负载均衡。


　　额。。。还没看“动手写WordCount”的兄弟姐妹们，建议先去看看。
　　▶ 数据位置感知
　　下面是WordCount的业务逻辑代码：
　　val file = "hdfs://127.0.0.1:9000/file.txt"
　　val lines = sc.textFile(file)
　　val words = lines.flatMap(line => line.split("\\s+"))
　　val wordCount = words.countByValue()
　　lines是Spark的RDD，它包含了在哪些机器上有file文件的块，信息是从HDFS来的。每文件块映射到RDD上就是一个分区，对的，没看错。如果一个文件块128MB，那么HDFS上一个1GB大小的文件就有8个文件块，由这个文件创建的RDD就会有8个分区。
　　之前说了，在HDFS上每个文件块默认会有3份，那RDD的分区选择了那一份呢?对滴，根据负载选择服务器负载最低的那一份。负载自动均衡了吧。


　　计算逻辑分发
　　有了这些信息，我们就知道把后续的计算逻辑该分发到哪儿去。
　　首先，我们得说清楚什么是计算逻辑，各位看官们想一下，类方法里面的代码是如何运行的。充分必要条件：方法代码 + 类实例(对象)的状态。似成相识吧，程序 = 算法 + 数据。算法在代码中，数据在对象的状态中。



　　Spark要分发计算逻辑，也是分了两部分。
　　第一部分是代码。为什么spark-submit执行一开始，总是一堆jar包被分发，原因就在这儿。
　　第二部分是类实例。类在哪儿?作为RDD各API参数的闭包。
　　val words = lines.flatMap(line => line.split("\\s+"))
　　flatMap的参数 **_.split("\s+")** 是闭包，闭包是引用了外部自由变量的函数，在Scala中是由匿名类实现的。更多信息，请小伙伴们GFSOSO哈。
　　上面的一行代码中，Spark要分发的实例就是 **_.split("\s+")** 的实例。
　　val wordCount = words.countByValue()
　　实际上RDD的API countByValue 也有需要分发的闭包实例，只是都在Spark的源码中，让一码给大家整理到明面上来哈。
　　val wordCount = words
　　.mapPartitions(convertWordsInPartitionToWordCountMap)
　　.reduce(mergeMaps)
　　前面我们提到了RDD的分区，mapPartitions会方法中的逻辑放到RDD的每个分区上执行，注意是远程在Slave上执行的哈。而reduce是在把每个分区的结果拿到Driver后，对结果进行两两合并，最终得到结果。
　　WordCount分布式运行原理

　　先仔细看图，相信不用下面的解释，各位看官也能看懂了。(上面的图是张巨高清的图，手机上看不清，建议转发文章到邮箱，然后到电脑上看，看懂这张图，就真的把WordCount分布式运行的机制搞懂了。)
　　对于WordCount而言，分布式在每个Slave的每个分区上，统计本分区内的单词计数，生成一个Map，然后将它传回给Driver，再由Driver两两合并来自各个分区的所有Map，形成最终的单词计数。
　　今天我们不仅说清楚了WordCount背后的分布式运行机制，而且解释了Spark的水平扩展能力，以及负载均衡。
　　下一篇将透过WordCount来看重中之重的容错处理，这涉及到Spark的应用场景与RDD的设计来源，可以毫不夸张地说，这才是Spark的精髓。


　　提示汇总
　　分布式处理系统会把计算逻辑分发到数据侧，极大提高系统的水平扩展性。