MapReduce学习笔记，理解学习Hadoop的MapReduce计算系统

Posted 2021-11-23 阿洋太爱大数据

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了MapReduce学习笔记，理解学习Hadoop的MapReduce计算系统相关的知识，希望对你有一定的参考价值。

MapReduce概述：

MapReduce最早是在Google的论文中提出的，但是对应的代码并没有开源。从2004年Google公开发布MapReduce论文到2012为止，MapReduce已经成长为被广泛采用的分布式数据处理的业界标准。

MapReduce是一种思想，总结也就是：“分而治之，迭代汇总”

MapReduce简介：

Hadoop MapReduce将作业分成一系列运行在分布式集群中的map任务和reduce任务。每个任务都工作在指定的小的数据子集上，因此负载是遍布集群中各个节点上的，其中：

map：负责数据的载入，解析，转换，过滤

reduce：负责处理map任务的输出结果的一个子集（reducer任务从mapper任务处2复制map任务的中间数据，进行聚合和分组操作）

map：

任务可以细分为：record reader、 mapper 、combiner、partitioner 四个阶段

map任务的输处被称为中间建和中间值，会发送到reducer做后续的处理。map任务运行的节点会优先选择数据所在的节点，因此，一般可以通过在本地机器上进行计算来减少数据的网络传输。

reduce：

任务可以细分为：混排、排序、reducer、输出格式四个阶段

MapReduce数据类型与格式：

1.内置数据类型：

hadoop内置数据类型主要有：

（1）BooleanWritable：标准布尔型数值

（2）ByteWritable：单字数值

（3）DoubleWritable：双字节数

（4）FloatWritable 浮点数

（5）IntWritable：整型数;

（6） LongWritable: 长整型数;

（7） Text：适用UTF8格式存储的文本

（8） NullWritable：null值;

（9）ArrayWritable：Writable类型的数组

2.用户自定义数据类型：

第一个基本要求：

实现writable接口，以便数据能被序列化后完成网络传输或文件输入\\输出

第二个基本要求：

若该数据需作为主键key使用or需比较值大小时，则需实现WritableComparable接口。

3.数据输入格式：（InputFormat）

用于描述MapReduce作业的数据输入规范，MapReduce框架依靠数据输入规范检查，对数据文件进行输入分块（InputSplit），及提供输入分块中将数据逐一读出并转换为map过程的输入键值对等功能。

最常用的数据输入格式包括：TextInputFormat

KeyValueTextInput

4.MapReduce架构：“分而治之”分解与结果的汇总

数据分片：

两个重要的函数：map和reduce函数

map：把任务分解成多个任务

reduce：负责把分解后的多任务处理的结果汇总起来

这两个函数的具体功能由用户根据自己的需求设计，只要按照用户自定义的规则，将输入的<key,value>对转换成另一个或一批<key,value>对输出即可。

理解：整体过程的目标就像是做果酱一样，map过程就是负责把需要的水果，分别切碎，这就是各自作用在这些物体上的一个Map操作。所以你给Map一个橘子，Map就会把它切碎。同样的，你把芒果，番茄一一地拿给Map，你也会得到各种碎块。所以，当你在切像果蔬时，你执行就是一个Map操作。Reduce（化简）:在这一阶段，你将各种果碎都放入研磨机里进行研磨，你就可以得到一瓶果酱了。这意味要制成一瓶果酱，你得研磨所有的原料。因此，研磨机通常将map操作的果蔬碎聚集在了一起。

MapReduce处理大数据集的过程：

在map阶段：

MapReduce框架将任务的输入数据分割成固定大小的片段（splits），随后将每个split进一步分解成批键值对<K1, V1>。Hadoop 为每一个 split 创建一个 map任务(以下简称Mapper)用于执行用户自定义的map函数，并将对应split中的<K1, V1>对作为输人，得到计算的中间结果<K2, V2>。接着将中间结果按照K2进行排序，并将key值相同的value放在一起形成一个新的表，形成<K2, list( V2)>元组。最后再根据key值的范围将这些元组进行分组，对应不同的reduce任务（简称reducer）

总的来说：其中collect和spill是属于shuffle阶段，而shuffle是横跨map和reduce阶段的（会在后文中提到）

   1. Read 阶段： MapTask 通过用户编写的 RecordReader ，从输入的 InputSplit 中解析出一个个 key / value 。
   2. Map 阶段：将解析出的 key / value 交给用户编写的 Map ()函数处理，并产生一系列新的 key / value 。
   3. Collect 阶段：在用户编写的 map() 函数中，数据处理完成后，一般会调用 outputCollector.collect() 输出结果，在该函数内部，它会将生成的 key / value 分片(通过调用 partitioner )，并写入一个环形内存缓冲区中(该缓冲区默认大小是 100MB )。
    4.Spill 阶段（此处的spill对应后问的sort（排序），溢写（spill）这两个过程）：即“溢写”，当缓冲区快要溢出时(默认达到缓冲区大小的 80 %)，会在本地文件系统创建一个溢出文件，将该缓冲区的数据写入这个文件。

    将数据写入本地磁盘前，先要对数据进行一次本地排序，并在必要时对数据进行合并、压缩等操作。
    写入磁盘之前，线程会根据 ReduceTask 的数量，将数据分区，一个 Reduce 任务对应一个分区的数据。
    这样做的目的是为了避免有些 Reduce 任务分配到大量数据，而有些 Reduce 任务分到很少的数据，甚至没有分到数据的尴尬局面。
    如果此时设置了 Combiner ，将排序后的结果进行 Combine 操作，这样做的目的是尽可能少地执行数据写入磁盘的操作。

5. Combine 阶段（对应后文的merge（合并））：当所有数据处理完成以后， MapTask 会对所有临时文件进行一次合并，以确保最终只会生成一个数据文件

    合并的过程中会不断地进行排序和 Combine 操作，
    其目的有两个：一是尽量减少每次写人磁盘的数据量;二是尽量减少下一复制阶段网络传输的数据量。
    最后合并成了一个已分区且已排序的文件。

在reduce阶段：

reducer把从不同mapper接受来的数据整合在一起，并进行排序，然后调用用户自定义reduce函数，对输入的<K2, list( V2)>对进行相应的处理，得到键值对<K3,V3>并输入到HDFS上。

总的来说：

      1. Copy 阶段（属于shuffle）： Reduce 会从各个 MapTask 上远程复制一片数据（每个 MapTask 传来的数据都是有序的），并针对某一片数据，如果其大小超过一定國值，则写到磁盘上，否则直接放到内存中
        2. Merge 阶段（属于shuffle）：在远程复制数据的同时， ReduceTask 会启动两个后台线程，分别对内存和磁盘上的文件进行合并，以防止内存使用过多或者磁盘文件过多。
        3.Sort 阶段（属于shuffle）：用户编写 reduce() 方法输入数据是按 key 进行聚集的一组数据。

为了将 key 相同的数据聚在一起， Hadoop 采用了基于排序的策略。
由于各个 MapTask 已经实现对自己的处理结果进行了局部排序，因此， ReduceTask 只需对所有数据进行一次归并排序即可。

4.Reduce 阶段：对排序后的键值对调用 reduce() 方法，键相等的键值对调用一次 reduce()方法，每次调用会产生零个或者多个键值对，最后把这些输出的键值对写入到 HDFS 中
5.Write 阶段： reduce() 函数将计算结果写到 HDFS 上。

合并的过程中会产生许多的中间文件(写入磁盘了)，但 MapReduce 会让写入磁盘的数据尽可能地少，并且最后一次合并的结果并没有写入磁盘，而是直接输入到 Reduce 函数。

Shuffle：

        MapReduce 工作过程中， Map 阶段处理的数据如何传递给 Reduce 阶段，这是 MapReduce 框架中关键的一个过程，这个过程叫作 Shuffle 。
       Shuffle 会将 MapTask 输出的处理结果数据分发给 ReduceTask ，并在分发的过程中，对数据按 key 进行分区和排序。
       Shuffle横跨map端和reduce端，在map端包括spill(溢写)，而在reduce端包括copy和sort过程

map端的spill：

“其中spill过程包括四个过程：collect（输出），sort（排序），溢写（spill），merge（合并）

Collect:

每个Map任务不断地以对的形式把数据输出到在内存中构造的一个环形数据结构中。使用环形数据结构是为了更有效地使用内存空间，在内存中放置尽可能多的数据。这个数据结构其实就是个字节数组，叫Kvbuffer

Sort:

先把Kvbuffer中的数据按照partition值和key两个关键字升序排序，移动的只是索引数据，排序结果是Kvmeta中数据按照partition为单位聚集在一起，同一partition内的按照key有序。

Spill:

Spill线程为这次Spill过程创建一个磁盘文件：从所有的本地目录中轮训查找能存储这么大空间的目录，找到之后在其中创建一个类似于 “spill12.out”的文件。Spill线程根据排过序的Kvmeta挨个partition的把数据吐到这个文件中，一个partition对应的数据吐完之后顺序地吐下个partition，直到把所有的partition遍历完。一个partition在文件中对应的数据也叫段(segment)。

内存缓冲区是有大小限制的，默认是100MB。当map task的输出结果很多时，就可能会撑爆内存，所以需要在一定条件下将缓冲区中的数据临时写入磁盘，然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为Spill，中文可译为溢写。比例默认是0.8，也就是当缓冲区的数据已经达到阈值（buffer size * spill percent = 100MB * 0.8 = 80MB），溢写线程启动，锁定这80MB的内存，执行溢写过程。Map task的输出结果还可以往剩下的20MB内存中写，互不影响。

Merge：

Map任务如果输出数据量很大，可能会进行好几次Spill，out文件和Index文件会产生很多，分布在不同的磁盘上。最后把这些文件进行合并的merge过程闪亮登场。如前面的例子，“aaa”从某个map task读取过来时值是5，从另外一个map 读取时值是8，因为它们有相同的key，所以得merge成group。什么是group。对于“aaa”就是像这样的：{“aaa”, [5, 8, 2, …]}

”

reduce的shuffle过程：

Copy：

Reduce 任务通过HTTP向各个Map任务拖取它所需要的数据。每个节点都会启动一个常驻的HTTP server，其中一项服务就是响应Reduce拖取Map数据。当有MapOutput的HTTP请求过来的时候，HTTP server就读取相应的Map输出文件中对应这个Reduce部分的数据通过网络流输出给Reduce。

Merge SORT：

这里使用的Merge和Map端使用的Merge过程一样。Map的输出数据已经是有序的，Merge进行一次合并排序，所谓Reduce端的 sort过程就是这个合并的过程。一般Reduce是一边copy一边sort，即copy和sort两个阶段是重叠而不是完全分开的。

当Reducer的输入文件已定，整个Shuffle才最终结束