cloudera learning8:MapReduce and Spark

Posted 2020-08-12 软件心理学工程师

YARN:Yet Another Resource Negotiator, Hadoop集群的资源管理器，可以对运行在Hadoop上的MapReduce V2，Spark，Impala等进行内存和CPU的分配。

MapReduce过程分析

input split（输入分片）阶段:在进行Map之前，会根据输入文件的大小进行输入分片，每个输入分片对应一个Map任务。输入分片本身并不记录输入数据的内容，而是记录一个分片的长度和一个记录数据所在位置的数组。输入分片往往和HDFS的block关系密切，比如，一个block的大小为128M，有三个输入文件，大小分别为3M，129M，254M则输入分片会把3M分为1个输入分片，129M分成两个分片，254M也会分成两个输入分片。如果我们在map之前做文件合并，把小文件合成大文件，就能够减少map的数量，可以优化MapReduce执行效率。

Map阶段：执行Map function。基于数据本地化原则，一般map的执行是在存储数据的DataNode节点上执行的。

Sort操作：sort是MapReduce模型默认的行为，根据Key对序列化的字节做的排序。 

Combiner操作：可选。Combiner实际上是DataNode节点(单个Map的数据)本地数据做的Reduce操作。先对Map结果进行局部Reduce，最大的好处是减少了Reduce阶段数据传输的大小。不过也不是所有的应用都适合进行局部Reduce，要根据业务的需要进去操作。

Merge操作：将spill的文件归并到一起。它是怎么做的呢？“aaa”从某个map task读取过来时值是5，从另外一个map 读取时值是8，因为它们有相同的key，所以得merge成group。什么是group。对于“aaa”就是像这样的：{“aaa”, [5, 8, 2, …]}，数组中的值就是从不同溢写文件中读取出来的，然后再把这些值加起来。

Shuffle阶段：将Map的输出作为Reduce的输入的过程就是shuffle了。Map结果输出时，会在内存里开启一个环形内存缓冲区，默认大小为100M。并且在配置文件里为这个缓冲区设定了一个阀值，默认是0.80，同时Map还会为输出操作启动一个守护线程，如果缓冲区的内存达到了阀值的80%时候，这个守护线程就会把内容写到磁盘上，这个过程叫spill，另外的20%内存可以继续写入要写进磁盘的数据，写磁盘和写内存是互不干扰的，如果内存区被撑满了，那么map久会阻塞写内存的操作，让写磁盘操作完成后再继续执行写内存，写磁盘之前，还会先后执行Sort和Combiner操作。写入磁盘的操作叫做spill，即内存满了往磁盘上“溢出”一个文件。也就是说在做map输出有几次spill就会产生多少个溢出文件，等map输出全部做完后，map会Merge这些溢出文件。这个过程里还有一个Partitioner，其实Partitioner操作和map阶段的输入分片（Input split）很像，一个Partitioner对象一个reduce作业，如果我们的mapreduce操作只有一个reduce操作，那么Partitioner就只有一个，如果有多个reduce操作，那么Partitioner对应会有多个，Partitioner就是reduce的输入分片，这个可以编程控制，主要是根据实际Key-Value的值，根据实际业务类型或者为了更好的reduce负载均衡要求进行，这是提高reduce效率的一个关键所在。到了reduce阶段就是合并map输出的文件，Partitioner会找到对应的map输出文件，然后进行复制操作（通过HTTP请求，把需要Reduce的数据聚到一起（HTTP传输也是影响performance的主要地方）），复制操作时reduce会开启几个复制线程，这些线程默认个数是5个，程序员也可以在配置文件更改复制线程的个数，这个复制过程和map写入磁盘过程类似，也有阀值和内存大小，阀值一样可以在配置文件里配置。 

Reduce阶段：执行reduce function，把结果写到HFDS上。

 

Spark过程分析

不同于Hadoop MapReduce把数据处理流程分成明显的input split，map，spill，merge，shuffle， sort，reduce等几个阶段，每个阶段各司其职，spark的核心是RDD，在执行阶段分为不同的stage和一系列transformation，action操作。Spark会记录所有的transformation操作，绘制一个DAG图，DAG中的RDDs之间如果是窄依赖，则化为同一个stage，如果是宽依赖，则分成不同的stage。

• 窄依赖是指父RDD的每个分区只被子RDD的一个分区所使用，子RDD分区通常对应常数个父RDD分区(O(1)，与数据规模无关)

• 相应的，宽依赖是指父RDD的每个分区都可能被多个子RDD分区所使用，子RDD分区通常对应所有的父RDD分区(O(n)，与数据规模有关)

Spark运行过程中记录所有的transformation操作（不实际执行），遇到Action操作，把所有之前的transformation操作一把进行执行。在执行的过程中，也可能遇到shuffle的case（比如，reduceByKey操作），spark的shuffle也是将mapper的输出通过Partitioner操作送入Reducer。Reducer以内存做缓冲，变shuffle边做aggerate，等到aggerate完成之后，执行reduce function。