MapReduce(mr) -- 分布式计算框架

Hadoop -YARN

Hadoop -HDFS

导读

一.MapReduce介绍

二.MapReduce工作流程及shuffle

三.MapReduce框架结构

四.MapReduce优化 -- 待完成

一. MapReduce介绍

mr百科

我们先看一下百科给的解释,可以看到已经给出了很明确的解释了
MapReduce思想 MapReduce思想在生活中处处可见。或多或少都曾接触过这种思想。MapReduce的思想核心是“分而治之”，适用于大量复杂的任务处理场景（大规模数据处理场景）。即使是发布过论文实现分布式计算的谷歌也只是实现了这种思想，而不是自己原创。
Map负责“分”，即把复杂的任务分解为若干个“简单的任务”来并行处理。可以进行拆分的前提是这些小任务可以并行计算，彼此间几乎没有依赖关系。
Reduce负责“合”，即对map阶段的结果进行全局汇总。
这两个阶段合起来正是MapReduce思想的体现。

mr模型

Hadoop MapReduce构思体现在如下的三个方面：
如何对付大数据处理：分而治之
对相互间不具有计算依赖关系的大数据，实现并行最自然的办法就是采取分而治之的策略。并行计算的第一个重要问题是如何划分计算任务或者计算数据以便对划分的子任务或数据块同时进行计算。不可分拆的计算任务或相互间有依赖关系的数据无法进行并行计算！
构建抽象模型：Map和Reduce
MapReduce借鉴了函数式语言中的思想，用Map和Reduce两个函数提供了高层的并行编程抽象模型。
Map: 对一组数据元素进行某种重复式的处理；
Reduce: 对Map的中间结果进行某种进一步的结果整理。
MapReduce中定义了如下的Map和Reduce两个抽象的编程接口，由用户去编程实现:
map: (k1; v1) → [(k2; v2)] , reduce: (k2; [v2]) → [(k3; v3)]
Map和Reduce为程序员提供了一个清晰的操作接口抽象描述。通过以上两个编程接口，可以看出MapReduce处理的数据类型是<key,value>键值对。
统一构架，隐藏系统层细节,如何提供统一的计算框架，如果没有统一封装底层细节，那么程序员则需要考虑诸如数据存储、划分、分发、结果收集、错误恢复等诸多细节；为此，MapReduce设计并提供了统一的计算框架，为程序员隐藏了绝大多数系统层面的处理细节。
MapReduce最大的亮点在于通过抽象模型和计算框架把需要做什么(what need to do)与具体怎么做(how to do)分开了，为开发人员提供一个抽象和高层的编程接口和框架。开发人员仅需要关心其应用层的具体计算问题，仅需编写少量的处理应用本身计算问题的程序代码。如何具体完成这个并行计算任务所相关的诸多系统层细节被隐藏起来,交给计算框架去处理：从分布代码的执行，到大到数千小到单个节点集群的自动调度使用。

总结一下MR就是 :
1.高可靠性,高扩展性,高效性,高容错性,低成本
2.MapReduce是一种分布式计算模型,是Google提出的,主要用于搜索领域,解决海量数据的计算问题,同时也是一种解决问题的思想,分而治之, 即先分后和,
3.MR有两个阶段组成：Map和Reduce,用户只需实现map()和reduce()两个函数,即可实现分布式计算;

二. MapReduce工作流程及shuffle

大体分为 input,map,shuffle,reduce,output,其中shuffle分为(map阶段shuffle(重点)和reduce阶段的shuffle),即 从 输入 -> map -> shuffle -> reduce -> 输出,其中最要步骤为shuffle阶段,会重点进行记录

1.input -- split && map

block块(物理划分)
block是HDFS中的基本存储单位,默认大小为128M,文件上传到HDFS,就要划分数据成块,这里的划分属于物理的划分。
plit分片(逻辑划分)
Hadoop中split划分属于逻辑上的划分,目的只是为了让map task更好地获取数据。split是通过hadoop中的InputFormat接口中的getSplit()方法得到的。那么,split的大小具体怎么得到呢？
首先介绍几个数据量：
totalSize：整个mapreduce job输入文件的总大小。
numSplits：来自job.getNumMapTasks(),即在job启动时用户利用 org.apache.hadoop.mapred.JobConf.setNumMapTasks(int n)设置的值,从方法的名称上看,是用于设置map的个数。但是,最终map的个数也就是split的个数并不一定取用户设置的这个值,用户设置的map个数值只是给最终的map个数一个提示,只是一个影响因素,而不是决定因素。
goalSize：totalSize/numSplits,即期望的split的大小,也就是每个mapper处理多少的数据。但也仅仅是期望。
minSize：split的最小值,该值可由两个途径设置：
1.通过子类重写方法protected void setMinSplitSize(long minSplitSize)进行设置。一般情况为1,特殊情况除外
2.通过配置文件中的mapred.min.split.size进行设置最终取两者中的最大值！
split计算公式：finalSplitSize=max(minSize,min(goalSize,blockSize))

map
首先将 HDFS 中的数据以 Split 方式作为map的输入，可将键和值的输入数据行转换为输出键/值：
map（key1，value）-> list <key2，value2>
也就是说，对于输入，它返回包含零个或多个（键，值）对的列表：
1.输出可以是与输入不同的key
2.输出可以有多个具有相同key的记录

2.shuffle

shuffle的本意是洗牌、混洗的意思,把一组有规则的数据尽量打乱成无规则的数据。而在MapReduce中,shuffle更像是洗牌的逆过程,指的是将map端的无规则输出按指定的规则“打乱”成具有一定规则的数据,以便reduce端接收处理。其在MapReduce中所处的工作阶段是map输出后到reduce接收前,具体可以分为map端和reduce端前后两个部分。在shuffle之前,MapReduce会对要处理的数据进行分片(split)操作,每一个分片对应一个MapTask任务。接下来map()函数会对每一个分片中的每一行数据进行处理得到键值对(key,value),此时得到的键值对又叫做“中间结果”。此后便进入shuffle阶段,由此可以看出shuffle阶段的作用是处理“中间结果”。

mr过程

这是官方对shuffle的过程的一个描述,下面是中文的的看着更容易理解一点

mr过程-中文

1.mapShuffle

1.Partition
对于map输出的每一个键值对，系统都会给定一个partition，partition值默认是通过计算key的hash值后对Reduce task的数量取模获得。如果一个键值对的partition值为1，意味着这个键值对会交给第一个Reducer处理。
我们知道每一个Reduce的输出都是有序的，但是将所有Reduce的输出合并到一起却并非是全局有序的，如果要做到全局有序，我们该怎么做呢？最简单的方式，只设置一个Reduce task，但是这样完全发挥不出集群的优势，而且能应对的数据量也很受限。最佳的方式是自己定义一个Partitioner，用输入数据的最大值除以系统Reduce task数量的商作为分割边界，也就是说分割数据的边界为此商的1倍、2倍至numPartitions-1倍，这样就能保证执行partition后的数据是整体有序的。
另一种需要我们自己定义一个Partitioner的情况是各个Reduce task处理的键值对数量极不平衡。对于某些数据集，由于很多不同的key的hash值都一样，导致这些键值对都被分给同一个Reducer处理，而其他的Reducer处理的键值对很少，从而拖延整个任务的进度。当然，编写自己的Partitioner必须要保证具有相同key值的键值对分发到同一个Reducer。
即在将map()函数处理后得到的（key,value）对写入到缓冲区之前，需要先进行分区操作，这样就能把map任务处理的结果发送给指定的reducer去执行，从而达到负载均衡，避免数据倾斜。MapReduce提供默认的分区类(HashPartitioner),其中主要方法getPartition()方法有三个参数，前两个指的是mapper任务输出的键值对，而第三个参数指的是设置的reduce任务的数量，默认值为1。因为任何整数与1相除的余数肯定是0。也就是说默认的getPartition()方法的返回值总是0，也就是Mapper任务的输出默认总是送给同一个Reducer任务，最终只能输出到一个文件中。如果想要让mapper输出的结果给多个reducer处理，那么只需要写一个类，让其继承Partitioner类，并重写getPartition()方法，让其针对不同情况返回不同数值即可。并在最后通过job设置指定分区类和reducer任务数量即可。

2.Collector
Map的输出结果是由collector处理的，每个Map任务不断地将键值对输出到在内存中构造的一个环形数据结构中。使用环形数据结构是为了更有效地使用内存空间，在内存中放置尽可能多的数据。
因为频繁的磁盘I/O操作会严重的降低效率，因此“中间结果”不会立马写入磁盘，而是优先存储到map节点的“环形内存缓冲区”，并做一些预排序以提高效率，当写入的数据量达到预先设置的阙值后便会执行一次I/O操作将数据写入到磁盘。每个map任务都会分配一个环形内存缓冲区，用于存储map任务输出的键值对（默认大小100MB，mapreduce.task.io.sort.mb调整）以及对应的partition，被缓冲的（key,value）对已经被序列化（为了写入磁盘）。

3.Sort&Spill
当Spill触发后，SortAndSpill先把环形内存缓冲区中的数据按照partition值和key两个关键字升序排序，移动的只是索引数据，排序结果是环形内存缓冲区中数据按照partition为单位聚集在一起，同一partition内的按照key有序。一旦缓冲区内容达到阈值（mapreduce.map.io.sort.spill.percent,默认0.80，或者80%），就会会锁定这80%的内存，并在每个分区中对其中的键值对按键进行sort排序，具体是将数据按照partition和key两个关键字进行排序，排序结果为缓冲区内的数据按照partition为单位聚集在一起，同一个partition内的数据按照key有序。排序完成后会创建一个溢出写文件（临时文件），然后开启一个后台线程把这部分数据以一个临时文件的方式溢出写（spill）到本地磁盘中（如果客户端自定义了Combiner（相当于map阶段的reduce），则会在分区排序后到溢写出前自动调用combiner，将相同的key的value相加，这样的好处就是减少溢写到磁盘的数据量。这个过程叫“合并”）。剩余的20%的内存在此期间可以继续写入map输出的键值对。溢出写过程按轮询方式将缓冲区中的内容写到mapreduce.cluster.local.dir属性指定的目录中。
​

4.Merge​
当一个map task处理的数据很大，以至于超过缓冲区设定内存时，就会生成多个spill文件。此时就需要对同一个map任务产生的多个spill文件进行归并生成最终的一个已分区且已排序的大文件。配置属性mapreduce.task.io.sort.factor控制着一次最多能合并多少流，默认值是10。这个过程包括排序和combine(合并 - 可选)，归并得到的文件内键值对有可能拥有相同的key，这个过程如果设置了Combiner，也会合并相同的key值的键值对。
溢写的文件归并完毕后，Map将删除所有的临时溢出写文件，并告知NodeManager任务已完成，只要其中一个MapTask完成，ReduceTask就开始复制它的输出（Copy阶段分区输出文件通过http的方式提供给reducer）
​

combiner
​在merger之前还有一个操作就是Combine(合并) -- 可选
每一个 map 都可能会产生大量的本地输出，Combiner 的作用就是对 map 端的输出先做一次合并，以减少在 map 和 reduce 节点之间的数据传输量，以提高网络IO 性能，是 MapReduce 的一种优化手段之一。
1.combiner 是 MR 程序中 Mapper 和 Reducer 之外的一种组件
2.父类就 Reducer
3.combiner 和 reducer 的区别在于运行的位置：combiner 是在每一个 maptask 所在的节点运行 Reducer 是接收全局所有 Mapper 的输出结果；
4.combiner 的意义就是对每一个 maptask 的输出进行局部汇总，以减小网络传输量
 
如果指定了Combiner，可能在两个地方被调用：
1.当为作业设置Combiner类后，缓存溢出线程将缓存存放到磁盘时，就会调用；
2.缓存溢出的数量超过mapreduce.map.combine.minspills（默认3）时，在缓存溢出文件合并的时候会调用
合并（Combine）和归并（Merge）的区别：
两个键值对<“a”,1>和<“a”,1>，如果合并，会得到<“a”,2>，如果归并，会得到<“a”,<1,1>>。
combine为了减少数据量, 注意 : combine需要注意场合,并不是所有mr任务都适用

5.压缩
在shuffle阶段，从map阶段输出的数据，都要通过网络拷贝，发送到reduce阶段，这一过程中，涉及到大量的网络IO，如果数据能够进行压缩，写磁盘时压缩map端的输出，因为这样会让写磁盘的速度更快，节约磁盘空间，并减少传给reducer的数据量。默认情况下，输出是不压缩的(将mapreduce.map.output.compress设置为true即可启动）

2.reduceShuffle

1.Copy
Reduce任务通过HTTP向各个Map任务拖取它所需要的数据。Map任务成功完成后，会通知父TaskTracker状态已经更新，TaskTracker进而通知JobTracker（这些通知在心跳机制中进行）。所以，对于指定作业来说，JobTracker能记录Map输出和TaskTracker的映射关系。Reduce会定期向JobTracker获取Map的输出位置，一旦拿到输出位置，Reduce任务就会从此输出对应的TaskTracker上复制输出到本地，而不会等到所有的Map任务结束。

2.Merge Sort
 Copy过来的数据会先放入内存缓冲区中，如果内存缓冲区中能放得下这次数据的话就直接把数据写到内存中，即内存到内存merge。Reduce要向每个Map去拖取数据，在内存中每个Map对应一块数据，当内存缓存区中存储的Map数据占用空间达到一定程度的时候，开始启动内存中merge，把内存中的数据merge输出到磁盘上一个文件中，即内存到磁盘merge。在将buffer中多个map输出合并写入磁盘之前，如果设置了Combiner，则会化简压缩合并的map输出。Reduce的内存缓冲区可通过mapred.job.shuffle.input.buffer.percent配置，默认是JVM的heap size的70%。内存到磁盘merge的启动门限可以通过mapred.job.shuffle.merge.percent配置，默认是66%。
​
当属于该reducer的map输出全部拷贝完成，则会在reducer上生成多个文件（如果拖取的所有map数据总量都没有内存缓冲区，则数据就只存在于内存中），这时开始执行合并操作，即磁盘到磁盘merge，Map的输出数据已经是有序的，Merge进行一次合并排序，所谓Reduce端的sort过程就是这个合并的过程。一般Reduce是一边copy一边sort，即copy和sort两个阶段是重叠而不是完全分开的。最终Reduce shuffle过程会输出一个整体有序的数据块。

3.reduce && output

提供了一个reduce转换来获取特定key的所有值，并减少输出生成的新列表。
减少(key2, list<value2>) -> list<value3>

当reduce任务启动时，它的输入分散在运行map任务的所有节点上的许多文件中。如果以分布式模式运行，则需要在copy阶段首先将这些文件复制到本地文件系统
一旦所有数据在本地可用，它就会在append阶段追加到一个文件中。然后对文件进行合并排序，以便给定key的键值对是连续的(排序阶段)。这使得实际的reduce操作变得简单 : 按顺序读取文件，并使用迭代器将值传递给reduce方法，迭代器读取输入文件，直到遇到下一个键值。最后，输出将由每个执行的reduce任务的一个输出文件组成。文件的格式可以用JobConf.setOutputFormat. 如果使用SequentialOutputFormat，则还必须指定输出键和值类。

mr的shuffle总结

MapReduce处理数据过程主要分成2个阶段：map阶段和reduce阶段。先执行map阶段，再执行reduce阶段。

1).Collect阶段：将MapTask的结果输出到默认大小为100M的环形缓冲区，保存的是key/value，Partition分区信息等。
2).Spill阶段：当内存中的数据量达到一定的阀值的时候，就会将数据写入本地磁盘，在将数据写入磁盘之前需要对数据进行一次排序的操作，如果配置了combiner，还会将有相同分区号和key的数据进行排序。
3).Merge阶段：把所有溢出的临时文件进行一次合并操作，以确保一个MapTask最终只产生一个中间数据文件,merge期间可以设置combine提高效率(需要根据实际情况使用)。
4).Copy阶段：ReduceTask启动Fetcher线程到已经完成MapTask的节点上复制一份属于自己的数据，这些数据默认会保存在内存的缓冲区中，当内存的缓冲区达到一定的阀值的时候，就会将数据写到磁盘之上。
5).Merge阶段：在ReduceTask远程复制数据的同时，会在后台开启两个线程对内存到本地的数据文件进行合并操作。
6).Sort阶段：在对数据进行合并的同时，会进行排序操作，由于MapTask阶段已经对数据进行了局部的排序，ReduceTask只需保证Copy的数据的最终整体有效性即可。
Shuffle中的缓冲区大小会影响到mapreduce程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快
缓冲区的大小可以通过参数调整, 参数：mapreduce.task.io.sort.mb 默认100M

三. MapReduce序列化和 计数器

 
序列化

序列化（Serialization）是指把结构化对象转化为字节流。
反序列化（Deserialization）是序列化的逆过程。把字节流转为结构化对象。 当要在进程间传递对象或持久化对象的时候，就需要序列化对象成字节流
反之当要将接收到或从磁盘读取的字节流转换为对象，就要进行反序列化。
Java 的序列化（Serializable）是一个重量级序列化框架，一个对象被序列化后，会附带很多额外的信息（各种校验信息，header，继承体系等），不便于在网络中高效传输；所以，hadoop 自己开发了一套序列化机制（Writable），精简，高效。不用像 java 对象类一样传输多层的父子关系，需要哪个属性就传输哪个属性值，大大的减少网络传输的开销。
Writable是Hadoop的序列化格式，hadoop定义了这样一个Writable接口。 一个类要支持可序列化只需实现这个接口即可。
另外Writable有一个子接口是WritableComparable，writableComparable是既可实现序列化，也可以对key进行比较，我们这里可以通过自定义key实现WritableComparable来实现我们的排序功能

 
计数器

计数器是收集作业统计信息的有效手段之一，用于质量控制或应用级统计。计数器还可辅助诊断系统故障。如果需要将日志信息传输到map 或reduce 任务， 更好的方法通常是看能否用一个计数器值来记录某一特定事件的发生。对于大型分布式作业而言，使用计数器更为方便。除了因为获取计数器值比输出日志更方便，还有根据计数器值统计特定事件的发生次数要比分析一堆日志文件容易得多。

每次mapreduce执行完成之后，我们都会看到一些日志记录出来，其中最重要的一些日志记录如下截图, 具体应用需要根据项目情况使用

image.png

三. MapReduce框架结构

一个完整的mapreduce程序在分布式运行时有三类实例进程：
1、MRAppMaster：负责整个程序的过程调度及状态协调
2、MapTask：负责map阶段的整个数据处理流程
3、ReduceTask：负责reduce阶段的整个数据处理流程

mr任务运行

map

详细步骤：
1、 首先，读取数据组件InputFormat（默认TextInputFormat）会通过getSplits方法对输入目录中文件进行逻辑切片规划得到splits，有多少个split就对应启动多少个MapTask。split与block的对应关系默认是一对一可以更改配置。
2、 将输入文件切分为splits之后，由RecordReader对象（默认LineRecordReader）进行读取，以\n作为分隔符，读取一行数据，返回<key，value>。Key表示每行首字符偏移值，value表示这一行文本内容。
3、 读取split返回<key,value>，进入用户自己继承的Mapper类中，执行用户重写的map函数。RecordReader读取一行这里调用一次。
4、 map逻辑完之后，将map的每条结果通过context.write进行collect数据收集。在collect中，会先对其进行分区处理，默认使用HashPartitioner。
MapReduce提供Partitioner接口，它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对Partitioner有需求，可以订制并设置到job上。
5、接下来，会将数据写入内存，内存中这片区域叫做环形缓冲区，缓冲区的作用是批量收集map结果，减少磁盘IO的影响。我们的key/value对以及Partition的结果都会被写入缓冲区。当然写入之前，key与value值都会被序列化成字节数组。
环形缓冲区其实是一个数组，数组中存放着key、value的序列化数据和key、value的元数据信息，包括partition、key的起始位置、value的起始位置以及value的长度。环形结构是一个抽象概念。
缓冲区是有大小限制，默认是100MB。当map task的输出结果很多时，就可能会撑爆内存，所以需要在一定条件下将缓冲区中的数据临时写入磁盘，然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为Spill，中文可译为溢写。这个溢写是由单独线程来完成，不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出，所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8，也就是当缓冲区的数据已经达到阈值（buffer size * spill percent = 100MB * 0.8 = 80MB），溢写线程启动，锁定这80MB的内存，执行溢写过程。Map task的输出结果还可以往剩下的20MB内存中写，互不影响。
6、当溢写线程启动后，需要对这80MB空间内的key做排序(Sort)。排序是MapReduce模型默认的行为，这里的排序也是对序列化的字节做的排序。
如果job设置过Combiner，那么现在就是使用Combiner的时候了。将有相同key的key/value对的value加起来，减少溢写到磁盘的数据量。Combiner会优化MapReduce的中间结果，所以它在整个模型中会多次使用。
那哪些场景才能使用Combiner呢？从这里分析，Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。比如累加，最大值等。Combiner的使用一定得慎重，如果用好，它对job执行效率有帮助，反之会影响reduce的最终结果。
7、合并溢写文件：每次溢写会在磁盘上生成一个临时文件（写之前判断是否有combiner），如果map的输出结果真的很大，有多次这样的溢写发生，磁盘上相应的就会有多个临时文件存在。当整个数据处理结束之后开始对磁盘中的临时文件进行merge合并，因为最终的文件只有一个，写入磁盘，并且为这个文件提供了一个索引文件，以记录每个reduce对应数据的偏移量。
至此map整个阶段结束。

mapTask的一些基础设置配置（mapred-site.xml当中）：
设置一：设置环型缓冲区的内存值大小（默认设置如下）
mapreduce.task.io.sort.mb 100
设置二：设置溢写百分比（默认设置如下）
mapreduce.map.sort.spill.percent 0.80
设置三：设置溢写数据目录（默认设置）
mapreduce.cluster.local.dir ${hadoop.tmp.dir}/mapred/local
设置四：设置一次最多合并多少个溢写文件（默认设置如下）
mapreduce.task.io.sort.factor 10

reduce

详细步骤：
1、Copy阶段，简单地拉取数据。Reduce进程启动一些数据copy线程(Fetcher)，通过HTTP方式请求maptask获取属于自己的文件。
2、Merge阶段。这里的merge如map端的merge动作，只是数组中存放的是不同map端copy来的数值。Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活。merge有三种形式：内存到内存；内存到磁盘；磁盘到磁盘。默认情况下第一种形式不启用。当内存中的数据量到达一定阈值，就启动内存到磁盘的merge。与map 端类似，这也是溢写的过程，这个过程中如果你设置有Combiner，也是会启用的，然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行，直到没有map端的数据时才结束，然后启动第三种磁盘到磁盘的merge方式生成最终的文件。
3、合并排序。把分散的数据合并成一个大的数据后，还会再对合并后的数据排序。
4、对排序后的键值对调用reduce方法，键相等的键值对调用一次reduce方法，每次调用会产生零个或者多个键值对，最后把这些输出的键值对写入到HDFS文件中。

四. MapReduce优化

Hadoop -YARN
Hadoop -HDFS
mr的优化还没完成