MapReduce整理

Driver阶段

​ 相当于YARN集群的客户端，用于提交我们整个程序到YARN集群，提交的是封装了MapReduce程序相关运行参数的job对象。

​ Java 的序列化是一个重量级序列化框架(Serializable)，对象被序列化后，会附带很多额外的信息(各种校验信息，Header，继承体系等)，网络传输效率低。所以Hadoop 自己开发了一套序列化机(Writable)。

Writable 序列化特点：

紧凑 ：高效使用存储。

快速：读写数据的额外开销小。

互操作：支持多语言的交互。

自定义bean对象序列化传输步骤及注意事项：

必须实现Writable接口

反序列化时，需要反射调用空参构造函数，所以必须有空参构造

重写序列化方法

@Overridepublic void write(DataOutput out) throws IOException {out.writeLong(id);out.writeString(name);out.writeLong(sumFlow);}

@Overridepublic void readFields(DataInput in) throws IOException {id = in.readLong; name = in.readString;}

注意反序列化方法中属性收集顺序和序列化方法中属性收集顺序完全一致

要想把结果显示在文件中，需要重写toString()，且用t分开，方便后续用

如果需要将自定义的bean放在key中传输，则还需要实现comparable接口，因为mapreduce框中的shuffle过程一定会对key进行排序

@Overridepublic int compareTo(FlowBean o) {// 倒序排列，从大到小return this.id > o.getId() ? -1 : 1;}

MapReduce 计算流程大概分为：

数据块：Block 是 HDFS 物理上把数据分成一块一块。数据块是 HDFS 存储数据单位。

数据切片：数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。数据切片是 MapReduce 程序计算输入数据的单位，一个切片会对应启动一个 MapTask，MapReduce默认使用TextFileInputFormat切片策略。

数据切片决定 MapTask 并发数量，这个 数据切片的默认值为 dfs.blocksize。这是由于 Yarn MrAppMaster在分配 MapTask 的时候，会优先考虑在存放Block的节点分配 MapTask，这样就能减少网络IO，提高效率。但是如果，数据切片大小不相等数据块大小，那么即使Block本地启动的MapTask也难免需要网络IO读取其他节点的Block，效率有所损失。

设置 ReduceTask 并行度（个数）

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

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

ReduceTask=0，表示没有Reduce阶段，输出文件个数和Map个数一致。

ReduceTask默认值就是1，所以输出文件个数为一个。

如果数据分布不均匀，就有可能在Reduce阶段产生数据倾斜

ReduceTask数量并不是任意设置，还要考虑业务逻辑需求，有些情况下，需要计算全局汇总结果，就只能有1个ReduceTask。

具体多少个ReduceTask，需要根据集群性能而定。

如果分区数不是1，但是ReduceTask为1，是否执行分区过程。答案是：不执行分区过程。因为在MapTask的源码中，执行分区的前提是先判断ReduceNum个数是否大于1。不大于1肯定不执行。

​ MapReduce任务提交之后，会生成一个 Yarn 客户端，提交任务到Yarn集群(本地环境会提交到本地)，提交流程如下：

MapReduce任务提交成功之后，就要开始执行任务了。执行任务第一步，当然是获取数据，然后启动MapTask，在这之前要先作切片处理了，下面是 FileInputFormat 切片机制的源码解析：

FileInputFormat 是 MapReduce 默认使用的切片机制，设计的核心思想是减少数据的网络IO传输，从而更高效的读取数据。有如下特点：

特点

简单地按照文件的内容长度进行切片

切片大小，默认等于Block大小

切片时不考虑数据集整体，而是逐个针对每一个文件单独切片

源码中计算切片大小的公式

// 默认值为1mapreduce.input.fileinputformat.split.minsize=1;// 默认值Long.MAXValuemapreduce.input.fileinputformat.split.maxsize= Long.MAXValue;// 默认情况下，切片大小=blocksize。 Math.max(minSize, Math.min(maxSize, blockSize));

maxsize（切片最大值）：参数如果比blockSize小，则会让切片变小，而且就等于配置的这个参数的值。

minsize（切片最小值）：参数调的比blockSize大，则可以让切片变得比blockSize还大。

TextInputFormat 其实是 FileInputFormat 默认的实现类，按行读取每条记录。键是存储该行在整个文件中的起始字节偏移量， LongWritable 类型。值是这行的内容，不包括任何行终止符（换行符和回车符），Text 类型。

框架默认的 TextInputFormat 切片机制是对任务按文件规划切片，不管文件多小，都会是一个单独的切片，都会交给一个 MapTask，这样如果有大量小文件，就会产生大量的MapTask，处理效率极其低下。

CombineTextInputFormat 用于小文件过多的场景，它可以将多个小文件从逻辑上规划到一个切片中，这样，多个小文件就可以交给一个 MapTask 处理。

下面是 CombineTextInputFormat 切片步骤：

Map阶段：

Reduce阶段：

准备待处理文件。

主要是提交任务资源，生成切片规划，详见[MapReduce任务提交](# 5.3、Job提交流程源码解析)。

将上一步生成的 Job.xml，Job.split(切片规划) 及 mr 程序/依赖 jar包 提交给 Yarn ResourceManager。

yarn mrAppMaster 接手调度 MapReduce程序，并通过 job.split 文件请求 NodeManger 启动 Container，在Container环境中，执行 MapReduce 程序。

MapReduce 程序第一步，获取数据，一般是从 HDFS 中去获取数据，切片详见 [ MapReduce 切片机制](#6、MapReduce 切片机制)。

执行 MapReduce 的逻辑代码，将计算的结果通过 outputCollector 写入到环形缓冲区。

7 ～ 15 为 Shuffle 操作，详见 [Shuffle机制](#7.1、Shuffle 机制)

数据处理完成后，进入 Reduce 方式，执行聚合逻辑，计算完成，使用 TextOutputFormat 写出结果到HDFS。

Map 方法之后，Reduce 方法之前的数据处理过程称之为 Shuffle。我们对Shuffle的期望是：将Map的输出数据完整地传输到Reduce端。 在传输数据时，尽可能得减少不必要的带宽消耗。 降低磁盘I/O的影响。步骤如下：

Map程序执行完成之后，结果先进入 getParitition( )，getParitition( )默认使用 hashPartition( )，即(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;。如果需要自定义分区规则，就要继承 Partition 抽象类，重写 getPartition( )，再任务中提交分区则Job.setPartitionClass( )，并且控制好 ReduceTask 数量Job.setNumReduceTasks()。为数据打上分区标记后，随后进入环形缓冲区。

环形缓冲区：
同时存储数据集 以及 数据集索引的内存区域，索引由 数据的index, partition, keyStart, valueStart组成，数据是由 key, value, unsued组成。 环形缓冲区 默认 100M，当数据存储空间达到环形缓冲区的 80% 时(为了在旧数据溢写时可以同时写入新数据)，环形缓冲区内的数据开始反写新数据 并 准备溢写旧数据(如果旧数据还没溢写完成，新数据已经占满内存，则新数据需要等待旧数据溢写完成，保证数据安全)。环形缓冲区的亮点在于，他可以同时兼顾 旧数据的溢写(持久化) 和 新数据的输入，并缺保证数据安全，效率相当高。

环形缓冲区旧数据准备溢写，溢写之前，将 环形缓冲区 中保存的索引默认按照分区 以及 key的字典顺序快排排序，如果想要自定义排序规则，则需要在key类中实现WritableComparable 接口重写 compareTo 方法，索引保存了数据的原地址，排序完成后修改索引保存的数据地址，就可以快速按序写入数据。

Combiner 是一个 独立与 Mapper 和 Reducer 之外，在 MapTask所在节点运行的以 Reducer 为父类的组件。由于 MapTask 数量远大于 ReduceTask 数量，所以 ReduceTask 成为了整个运行过程的性能短板，为了提高 ReduceTask 工作效率， MapReduce 提供 Combiner 组件，在 Map阶段结束，Shffle 过程中对 Map结果 先进行一次 归并计算，这样传输到 ReduceTask 的数据量大大减少，减少网络传输成本，并且由于计算数据量大大减少，那么 ReduceTask 计算时间也大大减少，实现性能提升。

但是，使用 Combiner 是有前提条件的，那就是不能影响最终的业务逻辑，而且，Combiner的输出kv应该跟Reducer的输入kv类型要对应。

自定义 Combiner 实现步骤：

自定义一个 Combiner 继承 Reducer，重写 Reduce 方法在驱动类中添加设置：job.setCombinerClass(WordCountCombiner.class);

按照分区编号由小到大依次将每个分区中的数据写入任务工作目录下的临时文件 output/spillN.out（N 表示当前溢写次数）中，将分区数据的元信息写到内存索引数据结构 SpillRecord 中，其中每个分区的元信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过 1MB，则将内存索引写到文件 output/spillN.out.index 中

溢写文件长期不处理，会比较杂乱，定期对溢写文件做 Merge 处理，同样是按照分区，排序的标准，对局部有序的数据排序，使用归并排序，提高效率。

由于上一步我们给多个溢写文件做了归并排序，合并后的大文件还可以进行一次 Combine，合并成一个更小的，计算后的文件。对 ReduceTask 即减少了 网络IO 时间，又减少了原始数据量，效率提高。

MapTask任务完成后，启动相应数量的ReduceTask，并告知ReduceTask处理数据范围(数据分区)。

ReduceTask 主动去 MapTask 中拉取Merge文件(要什么分区拉什么分区)，如果文件大小超过一定阈值，则溢写磁盘上，否则存储在内存中。ReduceTask 启动了两个后台线程，一个是负责监控内存中文件大小，超过一定阈值，则进行一次合并后将数据溢写到磁盘上。另一个负责监控磁盘上文件数目，达到一定阈值，则进行一次归并排序以生成一个更大文件；当所有数据拷贝完毕后，ReduceTask统一对内存和磁盘上的所有数据进行一次归并排序。

分区排序完成之后，还可以对数据进行分组。

数据分区可以帮助我们，根据业务场景来划分计算范围，完成符合需求的范围数据聚合统计等问题。而 MapReduce 的默认分区规则，使用了 hashPartition，源码如下：

public class HashPartitioner extends Partitioner {public int getPartition(K key, V value, int numReduceTasks) { return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks; } }

要想按照自己的方式去分区，就要继承 Partition 抽象类，重写 getPartition( )，再任务中提交分区规则Job.setPartitionClass( )，并且控制好 ReduceTask 数量Job.setNumReduceTasks()。

总结如下：