【Hadoop】MapReduce详解

MapReduce介绍

MapReduce的思想核心是“分而治之”，适用于大量复杂的任务处理场景（大规模数据处理场景）。

• Map负责“分”，即把复杂的任务分解为若干个“简单的任务”来并行处理。可以进行拆分的前提是这些小任务可以并行计算，彼此间几乎没有依赖关系。
• Reduce负责“合”，即对map阶段的结果进行全局汇总。

这两个阶段合起来正是MapReduce思想的体现。

MapReduce的基本编程模型

MapReduce是一个分布式运算程序的编程框架，核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在Hadoop集群上。

MapReduce设计并提供了统一的计算框架，为程序员隐藏了绝大多数系统层面的处理细节。为程序员提供一个抽象和高层的编程接口和框架。程序员仅需要关心其应用层的具体计算问题，仅需编写少量的处理应用本身计算问题的程序代码。如何具体完成这个并行计算任务所相关的诸多系统层细节被隐藏起来,交给计算框架去处理：

Map和Reduce为程序员提供了一个清晰的操作接口抽象描述。MapReduce中定义了如下的Map和Reduce两个抽象的编程接口，由用户去编程实现.Map和Reduce,MapReduce处理的数据类型是键值对。

• Map: (k1; v1) → [(k2; v2)]
• Reduce: (k2; [v2]) → [(k3; v3)]
  

一个完整的mapreduce程序在分布式运行时有三类实例进程：

• MRAppMaster 负责整个程序的过程调度及状态协调
• MapTask 负责map阶段的整个数据处理流程
• ReduceTask 负责reduce阶段的整个数据处理流程

在Hadoop中，用于执行MapReduce作业的机器角色有两个：JobTracker和TaskTracker。JobTracker用于调度作业，TaskTracker用于跟踪任务的执行情况。一个Hadoop集群只有一个JobTracker。

1）Client： 用户编写的MapReduce程序通过Client提交到JobTracker端，用户可通过Client提供的一些接口查看作业运行状态
2）JobTracker： JobTracker负责资源监控和作业调度， JobTracker 监控所有TaskTracker与Job的健康状况，一旦发现失败，就将相应的任务转移到其他节点， JobTracker 会跟踪任务的执行进度、资源使用量等信息，并将这些信息告诉任务调度器（TaskScheduler），而调度器会在资源出现空闲时，选择合适的任务去使用这些资源
3）TaskTracker：TaskTracker 会周期性地通过“心跳”将本节点上资源的使用情况和任务的运行进度汇报给JobTracker，同时接收JobTracker 发送过来的命令并执行相应的操作（如启动新任务、杀死任务等） TaskTracker 使用“slot”等量划分本节点上的资源量（CPU、内存等）。一个Task 获取到一个slot 后才有机会运行，而Hadoop调度器的作用就是将各个TaskTracker上的空闲slot分配给Task使用。slot 分为Map slot 和Reduce slot 两种，分别供MapTask 和Reduce Task 使用
4）Task： Task 分为Map Task 和Reduce Task 两种，均由TaskTracker 启动

MapReduce的计算过程

1. 拆分输入数据（Split）：系统会逐行读取文件的数据，得到一系列的（key/value）其中的key是一个偏移量，包括回车符在内的字符数
2. 执行Map方法：系统会将分割好的（key/value）对交给用户定义的Map方法进行处理，生成新的（key/value）
3. 排序和合并处理：系统在地道道Map方法输出的（key/value）对后，Mapper会将它们按照key值进行排序，并执行Combine过程，将key值相同的value值累加，得到Mapper的最终输出结果
4. Reduce阶段的排序与合并：Reduce先对Mapper接收的数据进行排序，再交由用户自定义的Reduce方法进行处理，得到新的（key/value）对。
   
   更细节的来看：

1. Map阶段可以概括为5个步骤：

①Read：Map Task通过用户编写的RecordReader，从输入InputSplit中解析出一个个的（key/value）。
②Map：该步骤主要将解析出的（key/value）交给用户编写的Map函数处理，并产生一些列新的（key/value）。
③Collect：在用户编写的Map函数中，数据处理完成后，一般会调用OutputCollector.collect()收集结果。在该函数内部，它将会生成（key/value）分片（通过Partitioner），并写入一个环形缓冲区中。
④Spill：即所谓溢写，指当环形缓冲区填满后，MapReduce会将数据写到本地磁盘上，生成一个临时文件。将数据写入本地磁盘之前，先要对数据进行一次本地排序，并在必要时对数据进行合并，压缩等操作。
⑤Combine：当所有数据处理完成后，Map Task对所有临时变量进行一次合并，以确保最终只会生成一个数据文件。

2. Reduce节点也可以分为5个步骤：

①Shuffle：也成为Copy阶段。Reduce Task从各个Map Task上远程复制一片数据，并针对某一片数据进行判断，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。
②Merge：在远程复制的同时，Reduce Task启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或者磁盘上文件过多。
③Sort：按照MapReduce语义，用户编写的Reduce函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚集在一起，Hadoop采用了基于排序的此策略。由于各个Map Task已经实现了对自己的数据结果进行了局部排序，因此，Reduce Task只需要对所有数据进行一次归并排序即可。
④Reduce：在该阶段中，Reduce Task将每组数据依次交给用户编写的reduce()函数处理
⑤Write：reduce()函数将计算结果写到HDFS。

设置ReduceTask并行度（个数）

ReduceTask的并行度同样影响整个Job的执行并发度和执行效率，但与MapTask的并发数由切片数决定不同，ReduceTask数量的决定是可以直接手动设置：

// 默认值是1，手动设置为4
job.setNumReduceTasks(4);
1
2

关于分片（Split）

HDFS 以固定大小的block 为基本单位存储数据，而对于MapReduce 而言，其处理单位是split。split 是一个逻辑概念，它只包含一些元数据信息，比如数据起始位置、数据长度、数据所在节点等。它的划分方法完全由用户自己决定。

• Map任务的数量：Hadoop为每个split创建一个Map任务，split 的多少决定了Map任务的数目。大多数情况下，理想的分片大小是一个HDFS块
• Reduce任务的数量：最优的Reduce任务个数取决于集群中可用的reduce任务槽(slot)的数目 通常设置比reduce任务槽数目稍微小一些的Reduce任务个数（这样可以预留一些系统资源处理可能发生的错误）
  

关于Shuffle

shuffle在map端和reduce都参与操作，所以可以分为map shuffle和reduce shuffle两个过程：

Map端的shuffle

从map的输出，需要经过分区、排序、合并过程输出为一个分区有序的文件。

首先根据数据的key值进行分区（默认是hash分区），然后数据写入一个环形缓冲区中，环形缓冲区的实质是一个字节数组，里面包含两部分数据，分别是数据和索引，索引中记录了每个key-value数据的分区等信息，环形缓冲区为100M，默认达到0.8的内存数据量时开始spill溢出到磁盘中，溢出过程前会对环形缓冲区按照partition和key进行排序操作（一般采用快排），也就是数据分区聚集，分区内按照key升序排列，如果这个时候设置了combiner的话，会按照相同的key进行合并，然后溢写到磁盘的一个文件中，当数据量很大的时候，会有多个这样的溢出小文件，多个小文件会按照分区进行合并，从而得到一个大的按照分区排序的输出文件。这是map shuffle做的事。

总结起来map shuffle需要做的事情有：
①分区partition（分区在进入缓冲区之前进行）
②写入环形内存缓冲区
③执行溢出写
排序sort（快排）—>合并combiner—>生成溢出写文件
④归并merge（归并排序），还可能再调用一次combiner

Reduce端的Shuffle

reduce shuffle也有两个过程，分别是复制map数据然后排序合并。

当map输出文件后，会将map输出和机器位置的映射信息报告给application master，同时reduce也会定期向application master询问，获得所需要复制数据的位置。reduce通过http从map端复制相应的数据到自己的内存缓冲区中，当内存数据量达到一定量的时候，进行merge合并，如果设置了combiner，还会combine操作，因为每个map文件已经是有序的，所以多个文件合并的时候采用的是根据key进行归并排序，这样reduce shuffle就产生了一个整体有序的数据块。

1. Reduce任务通过RPC向JobTracker询问Map任务是否已经完成，若完成，则领取数据
2. Reduce领取数据先放入缓存，来自不同Map机器，先归并，再合并，写入磁盘
3. 多个溢写文件归并成一个或多个大文件，文件中的键值对是排序的
4. 当数据很少时，不需要溢写到磁盘，直接在缓存中归并，然后输出给Reduce
   

Shuffle流程详解

首先看一下MapReduce中的排序的总体流程。
MapReduce框架会确保每一个Reducer的输入都是按Key进行排序的。一般，将排序以及Map的输出传输到Reduce的过程称为混洗（shuffle)。每一个Map都包含一个环形的缓存，默认100M，Map首先将输出写到缓存当中。当缓存的内容达到“阈值”时（阈值默认的大小是缓存的80%），一个后台线程负责将结果写到硬盘，这个过程称为“spill”。Spill过程中，Map仍可以向缓存写入结果，如果缓存已经写满，那么Map进行等待。

**Spill的具体过程如下：**首先，后台线程根据Reducer的个数将输出结果进行分组，每一个分组对应一个Reducer。其次，对于每一个分组后台线程对输出结果的Key进行排序。在排序过程中，如果有Combiner函数，则对排序结果进行Combiner函数进行调用。每一次spill都会在硬盘产生一个spill文件。因此，一个Map task有可能会产生多个spill文件，当Map写出最后一个输出时，会将所有的spill文件进行合并与排序，输出最终的结果文件。在这个过程中Combiner函数仍然会被调用。从整个过程来看，Combiner函数的调用次数是不确定的。

下面重点分析下Shuffle阶段的排序过程：
Shuffle阶段的排序可以理解成两部分，一个是对spill进行分区时，由于一个分区包含多个key值，所以要对分区内的<key,value>按照key进行排序，即key值相同的一串<key,value>存放在一起，这样一个partition内按照key值整体有序了。

第二部分并不是排序，而是进行merge，merge有两次，一次是map端将多个spill 按照分区和分区内的key进行merge，形成一个大的文件。第二次merge是在reduce端，进入同一个reduce的多个map的输出 merge在一起，该merge理解起来有点复杂，最终不是形成一个大文件，而且期间数据在内存和磁盘上都有。所以shuffle阶段的merge并不是严格的排序意义，只是将多个整体有序的文件merge成一个大的文件，由于不同的task执行map的输出会有所不同，所以merge后的结果不是每次都相同，不过还是严格要求按照分区划分，同时每个分区内的具有相同key的<key,value>对挨在一起。

Shuffle排序综述：如果只定义了map函数，没有定义reduce函数，那么输入数据经过shuffle的排序后，结果为key值相同的输出挨在一起，且key值小的一定在前面，这样整体来看key值有序（宏观意义的，不一定是按从大到小，因为如果采用默认的HashPartitioner，则key 的hash值相等的在一个分区，如果key为IntWritable的话，每个分区内的key会排序好的），而每个key对应的value不是有序的。

补充问题：

• Shuffle如何获取map输出数据给reduce？
  答：map执行完后会将map输出和机器位置的映射关系报告给application master，同时reduce会定期向application master询问，获得所需要的数据位置信息，之后reduce会通过http从map端复制相应的数据到reduce端的内存缓冲区中。

• Shuffle缺点：
  主要是数据传输IO的问题。由于从环形缓冲区需要溢写多个小文件到磁盘，产生较多的磁盘IO。

• combiner函数作用和作用在哪些地方？
  主要是实现本地key的聚合。一个map都可能会产生大量的本地输出，这些输出会通过网络到达reducer端，这样会浪费带宽。解决这个问题可以通过Combiner。Combiner的作用就是对map端的输出先做一次合并。
  combiner有三处：（1）从环形缓冲区写到磁盘（2）小文件合并为大文件（3）reduce的内存缓冲区溢写到磁盘中。

• mapreduce执行速度太慢，优化措施有哪些？：

1. 自定义分区（自己定义分割点，数据量多的时候采样排序确定分割点），让key值尽量均匀分布在每个分区上。
2. 当map端数据量太大的时候，可以对数据进行压缩。
3. 在不影响最后结果的情况下，使用combiner，在本地对key进行聚合。

MapReduce分区相关问题理解

在 MapReduce 中, 通过我们指定分区, 会将同一个分区的数据发送到同一个 Reduce 当中进行处理
例如: 为了数据的统计, 可以把一批类似的数据发送到同一个 Reduce 当中, 在同一个 Reduce 当中统计相同类型的数据, 就可以实现类似的数据分区和统计等。
其实就是相同类型的数据，有共性的数据, 送到一起去处理。默认的分区只有1个分区

1.Partition的原理和作用

得到map给的记录后，它们该分配给哪些reducer来处理呢？
hadoop采用的默认的派发方式是根据散列值来派发的，但是实际中，这并不能很高效或者按照我们要求的去执行任务。例如，经过partition处理后，一个节点的reducer分配到了20条记录，另一个却分配道了10W万条，试想，这种情况效率如何。又或者，我们想要处理后得到的文件按照一定的规律进行输出，假设有两个reducer，我们想要最终结果中part-00000中存储的是"h"开头的记录的结果,part-00001中存储其他开头的结果，这些默认的partitioner是做不到的。所以需要我们自己定制partition来根据自己的要求，选择记录的reducer。自定义partitioner很简单，只要自定义一个类，并且继承Partitioner类，重写其getPartition方法就好了，在使用的时候通过调用Job setPartitionerClass指定一下即可

Map的结果，会通过partition分发到Reducer上。Mapper的结果，可能送到Combiner做合并，Combiner在系统中并没有自己的基类，而是用Reducer作为Combiner的基类，他们对外的功能是一样的，只是使用的位置和使用时的上下文不太一样而已。Mapper最终处理的键值对<key, value>，是需要送到Reducer去合并的，合并的时候，有相同key的键/值对会送到同一个Reducer。哪个key到哪个Reducer的分配过程，是由Partitioner规定的。它只有一个方法，

getPartition(Text key, Text value, int numPartitions)
1

输入是Map的结果对<key, value>和Reducer的数目，输出则是分配的Reducer（整数编号）。就是指定Mappr输出的键值对到哪一个reducer上去。系统缺省的Partitioner是HashPartitioner，它以key的Hash值对Reducer的数目取模，得到对应的Reducer。这样保证如果有相同的key值，肯定被分配到同一个reducre上。如果有N个reducer，编号就为0,1,2,3……(N-1)。

2.Partition的使用

分区出现的必要性，如何使用Hadoop产生一个全局排序的文件？
最简单的方法就是使用一个分区，但是该方法在处理大型文件时效率极低，因为一台机器必须处理所有输出文件，从而完全丧失了MapReduce所提供的并行架构的优势。
事实上我们可以这样做，**首先创建一系列排好序的文件；其次，串联这些文件（类似于归并排序）；最后得到一个全局有序的文件。**主要的思路是使用一个partitioner来描述全局排序的输出。比方说我们有1000个1-10000的数据，跑10个ruduce任务， 如果我们运行进行partition的时候，能够将在1-1000中数据的分配到第一个reduce中，1001-2000的数据分配到第二个reduce中，以此类推。即第n个reduce所分配到的数据全部大于第n-1个reduce中的数据。这样，每个reduce出来之后都是有序的了，我们只要cat所有的输出文件，变成一个大的文件，就都是有序的了

基本思路就是这样，但是现在有一个问题，就是数据的区间如何划分，在数据量大，还有我们并不清楚数据分布的情况下。一个比较简单的方法就是采样，假如有一亿的数据，我们可以对数据进行采样，如取10000个数据采样，然后对采样数据分区间。在Hadoop中，patition我们可以用TotalOrderPartitioner替换默认的分区。然后将采样的结果传给他，就可以实现我们想要的分区。在采样时，我们可以使用hadoop的几种采样工具，RandomSampler,InputSampler,IntervalSampler。

这样，我们就可以对利用分布式文件系统进行大数据量的排序了，我们也可以重写Partitioner类中的compare函数，来定义比较的规则，从而可以实现字符串或其他非数字类型的排序，也可以实现二次排序乃至多次排序。

3.分组的概念和使用

分区的目的是根据Key值决定Mapper的输出记录被送到哪一个Reducer上去处理。而分组就是与记录的Key相关。在同一个分区里面，具有相同Key值的记录是属于同一个分组的。

GroupingComparator是mapreduce当中reduce端的一个功能组件，主要的作用是决定哪些数据作为一组，调用一次reduce的逻辑，默认是每个不同的key，作为多个不同的组，每个组调用一次reduce逻辑，我们可以自定义GroupingComparator实现不同的key作为同一个组，调用一次reduce逻辑.

分组排序的步骤

1. 自定义类继承WritableComparator
2. 重写compare()方法

@Override
public int compare(WritableComparable a, WritableComparable b) {
        // 比较的业务逻辑
        return result;
}
1
2
3
4
5

3. 创建一个构造将比较对象的类传给父类

protected OrderGroupingComparator() {
        super(OrderBean.class, true);
}
1
2
3

4.Combiner的使用

很多MapReduce程序受限于集群上可用的带宽，所以它会尽力最小化需要在map和reduce任务之间传输的中间数据。Hadoop允许用户声明一个combiner function来处理map的输出，同时把自己对map的处理结果作为reduce的输入。因为combiner function本身只是一种优化，hadoop并不保证对于某个map输出，这个方法会被调用多少次。换句话说，不管combiner function被调用多少次，对应的reduce输出结果都应该是一样的。

概念

每一个 map 都可能会产生大量的本地输出，Combiner 的作用就是对 map 端的输出先做一次合并，以减少在 map 和 reduce 节点之间的数据传输量，以提高网络IO 性能，是 MapReduce 的一种优化手段之一

• combiner 是 MR 程序中 Mapper 和 Reducer 之外的一种组件
• combiner 组件的父类就是 Reducer
• combiner 和 reducer 的区别在于运行的位置
  • Combiner 是在每一个 maptask 所在的节点运行
  • Reducer 是接收全局所有 Mapper 的输出结果
• combiner 的意义就是对每一个 map task 的输出进行局部汇总，以减小网络传输量

实现步骤

1. 自定义一个 combiner 继承 Reducer，重写 reduce 方法
2. 在 job 中设置 job.setCombinerClass(CustomCombiner.class)
   combiner 能够应用的前提是不能影响最终的业务逻辑，而且，combiner 的输出 k,v 应该跟 reducer 的输入 k,v 类型要对应起来

MapReduce排序和序列化

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

MapReduce简略步骤

• 第一步：读取文件，解析成为key，value对
• 第二步：自定义map逻辑接受k1,v1，转换成为新的k2,v2输出；写入环形缓冲区
• 第三步：分区：写入环形缓冲区的过程，会给每个k,v加上分区Partition index。（同一分区的数据，将来会被发送到同一个reduce里面去）
• 第四步：排序：当缓冲区使用80%，开始溢写文件
  • 先按partition进行排序，相同分区的数据汇聚到一起；
  • 然后，每个分区中的数据，再按key进行排序
• 第五步：combiner。调优过程，对数据进行map阶段的合并（注意：并非所有mr都适合combine）
• 第六步：将环形缓冲区的数据进行溢写到本地磁盘小文件
• 第七步：归并排序，对本地磁盘溢写小文件进行归并排序
• 第八步：等待reduceTask启动线程来进行拉取数据
• 第九步：reduceTask启动线程，从各map task拉取属于自己分区的数据
• 第十步：从mapTask拉取回来的数据继续进行归并排序
• 第十一步：进行groupingComparator分组操作
• 第十二步：调用reduce逻辑，写出数据
• 第十三步：通过outputFormat进行数据输出，写到文件，一个reduceTask对应一个结果文件