Hadoop之MapReduce详解

1. MapReduce概述

1.1 MapReduce定义

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

1.2 MapReduce优缺点

1.2.1 优点

1. MapReduce易于编程

它简单的实现一些接口，就可以完成一个分布式程序，这个分布式程序可以分布到大量廉价的PC机器上运行。也就是说你写一个分布式程序，跟写一个简单的串行程序是一模一样的。就是因为这个特点使得MapReduce编程变得非常流行。

1. 良好的扩展性

当你的计算资源不能得到满足的时候，你可以通过简单的增加机器来扩展它的计算能力。

1. 高容错性

MapReduce设计的初衷就是使程序能够部署在廉价的PC机器上，这就要求它具有很高的容错性。比如其中一台机器挂了，它可以把上面的计算任务转移到另外一个节点上运行，不至于这个任务运行失败，而且这个过程不需要人工参与，而完全是由Hadoop内部完成的。

1. 适合PB级以上海量数据的离线处理

可以实现上千台服务器集群并发工作，提供数据处理能力。

1.2.2 缺点

1. 不擅长实时计算

MapReduce无法像MySQL一样，在毫秒或者秒级内返回结果。

1. 不擅长流式计算

流式计算的输入数据是动态的，而MapReduce的输入数据集是静态的，不能动态变化。这是因为MapReduce自身的设计特点决定了数据源必须是静态的。

1. 不擅长DAG（有向无环图）计算

多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下，MapReduce并不是不能做，而是使用后，每个MapReduce作业的输出结果都会写入到磁盘，会造成大量的磁盘IO，导致性能非常的低下。

1.3 MapReduce核心思想

1. 分布式的运算程序往往需要分成至少2个阶段。
   1. 第一个阶段的MapTask并发实例，完全并行运行，互不相干。
   2. 第二个阶段的ReduceTask并发实例互不相干，但是他们的数据依赖于上一个阶段的所有MapTask并发实例的输出。

2. MapReduce编程模型只能包含一个Map阶段和一个Reduce阶段，如果用户的业务逻辑非常复杂，那就只能多个MapReduce程序，串行运行。

   1.4 MapReduce进程

   一个完整的MapReduce程序在分布式运行时有三类实例进程：
   （1）MrAppMaster：负责整个程序的过程调度及状态协调。
   （2）MapTask：负责Map阶段的整个数据处理流程。
   （3）ReduceTask：负责Reduce阶段的整个数据处理流程。

   1.5 MapReduce数据处理过程

   

3. Splitting阶段：输入分片，在Map计算之前，MApReduce会根据文件的大小进行数据的分片，每个分片针对一个Map任务，输入分片存储的并非数据本身，而是一个分片长度和一个记录数据的位置的数组。

4. Map阶段：Map一定有，Map操作都是本地化操作，也就是在数据存储节点上进行，并行的处理数据。
5. Combine阶段：也是并行处理数据，是一个局部统计的过程，此过程可有可无，主要作用是数据传输到reduce之前做一个数据整合，可以减少数据的传输量，以及reduce对数据的计算量，可以使数据的传输速度更快。
6. Shuffle阶段：把数据由一台机器的节点转到另一台机器节点的过程，把Map的输出结果转到reduce的过程，在此过程会根据数据的key进行排序，由于数据的传输会消耗性能，所以在开发的过程中要尽量避免shuffle。
7. Reduce阶段：对Map处理出的数据进行汇总，可以有多个reduce并行运行，一个reduce会产生一个输出文件，最终把结果存储在hdfs上。

   1.6 常用数据序列化类型

   | Java类型 | Hadoop Writable类型 | | —- | —- | | Boolean | BooleanWritable | | Byte | ByteWritable | | Int | IntWritable | | Float | FloatWritable | | Long | LongWritable | | Double | DoubleWritable | | String | Text | | Map | MapWritable | | Array | ArrayWritable | | Null | NullWritable |

2. MapReduce框架原理

2.1 InputFormat数据输入

2.1.1 切片与MapTask并行度决定机制

1）问题引出
MapTask的并行度决定Map阶段的任务处理并发度，进而影响到整个Job的处理速度。
思考：1G的数据，启动8个MapTask，可以提高集群的并发处理能力。那么1K的数据，也启动8个MapTask，会提高集群性能吗？MapTask并行任务是否越多越好呢？哪些因素影响了MapTask并行度？
2）MapTask并行度决定机制
数据块：Block是HDFS物理上把数据分成一块一块。数据块是HDFS存储数据单位。默认128M一块。
数据切片：数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。数据切片是MapReduce程序计算输入数据的单位，一个切片会对应启动一个MapTask。

2.1.2 Job提交流程

• Job提交流程源码 ```java waitForCompletion()

submit();

// 1建立连接 connect();
// 1）创建提交Job的代理 new Cluster(getConfiguration()); // （1）判断是本地运行环境还是yarn集群运行环境 initialize(jobTrackAddr, conf);

// 2 提交job submitter.submitJobInternal(Job.this, cluster)

// 1）创建给集群提交数据的Stag路径 Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);

// 2）获取jobid ，并创建Job路径 JobID jobId = submitClient.getNewJobID();

// 3）拷贝jar包到集群 copyAndConfigureFiles(job, submitJobDir);
rUploader.uploadFiles(job, jobSubmitDir);

// 4）计算切片，生成切片规划文件 writeSplits(job, submitJobDir); maps = writeNewSplits(job, jobSubmitDir); input.getSplits(job);

// 5）向Stag路径写XML配置文件 writeConf(conf, submitJobFile); conf.writeXml(out);

// 6）提交Job,返回提交状态 status = submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());

![image.png](https://cdn.nlark.com/yuque/0/2022/png/22837646/1651570880604-fe81fe47-c890-4d6c-9ee0-0334d38c707d.png#clientId=u2397d2ca-844c-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=389&id=u92f1b000&margin=%5Bobject%20Object%5D&name=image.png&originHeight=584&originWidth=1130&originalType=binary&ratio=1&rotation=0&showTitle=false&size=57131&status=done&style=stroke&taskId=uefea9131-e49f-49e1-b770-72e7543870a&title=&width=753.3333333333334)
<a name="my6qN"></a>
### 2.2 FileInputFormat切片机制
1. 一个Job的Map阶段并行度由客户端在提交Job时的切片数决定
1. 每一个Split切片分配一个MapTask并行实例处理
1. 默认情况下，切片大小=BlockSize
1. 切片是不考虑数据集整体，而是逐个针对每一个文件单独切片
<a name="KwHB6"></a>
#### 2.2.1 切片源码解析
1. MR程序先找到数据存储的目录
1. 开始遍历处理目录下额每一个文件
1. 遍历文件：
   1. 获取文件大小
   1. 计算切片大小
>  computeSplitSize（Math_max(minSize，Math_min(maxSize，blocksize))）=blocksize=128MB minSize默认1，maxSize默认Long最大值
> maxSize（切片最大值）：参数如果调的比blockSize小，则会让切片变小，而且就等于配置的这个参数的
> minSize（切片最小值）：参数调的比blockSize大，则可以让切片变得比blockSize还大
   3. 默认情况下，切片大小=blockSize（本地默认32MB，集群128MB）
   3. 开始切片。每次切片时，都要判断切完剩下的部分是否大于块的1.1倍，不大于1.1倍就划分为一块切片
   3. 将切片信息写到一个切片规划文件中
   3. 整个切片的核心在getSplit()方法中完成
   3. InputSplit只记录了切片的元数据信息，比如起始位置、长度以及所在的节点列表等
4. 提交切片规划文件到YARN上，YARN上的MrAppMaster就可以根据切片规划文件计算开启的MapTask个数。
<a name="QAZBE"></a>
#### 2.2.2 TextInputFormat
1. **FileInputFormat实现类**
思考：在运行MapReduce程序时，输入的文件格式包括：基于行的日志文件、二进制格式文件、数据库表等。那么，针对不同的数据类型，MapReduce是如何读取这些数据的呢？<br />FileInputFormat常见的接口实现类包括：TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat和自定义InputFormat等。
2. **TextInputFormat**
TextInputFormat是默认的FileInputFormat实现类。按行读取每条记录。键是存储该行在整个文件中的起始字节偏移量， LongWritable类型。值是这行的内容，不包括任何行终止符（换行符和回车符），Text类型。<br />**缺点**：不管文件多小，都会是一个单独的切片，都会交给一个MapTask处理，这样如果有大量小文件，就会产生大量的MapTask，处理效率极其低下。
<a name="ZkkBQ"></a>
#### 2.2.3 CombineTextInputFormat
1. **应用场景：**
**CombineTextInputFormat用于小文件过多的场景**，它可以将多个小文件从逻辑上规划到一个切片中，这样，多个小文件就可以交给一个MapTask处理。
2. **虚拟存储切片最大值设置**
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);  // 4m<br />注意：虚拟存储切片最大值设置最好根据实际的小文件大小情况来设置具体的值。
3. **切片机制**
生成切片过程包括：虚拟存储过程和切片过程二部分。<br />（1）虚拟存储过程：<br />将输入目录下所有文件大小，依次和设置的setMaxInputSplitSize值比较，如果不大于设置的最大值，逻辑上划分一个块。如果输入文件大于设置的最大值且大于两倍，那么以最大值切割一块；当剩余数据大小超过设置的最大值且不大于最大值2倍，此时将文件均分成2个虚拟存储块（防止出现太小切片）。<br />例如setMaxInputSplitSize值为4M，输入文件大小为8.02M，则先逻辑上分成一个4M。剩余的大小为4.02M，如果按照4M逻辑划分，就会出现0.02M的小的虚拟存储文件，所以将剩余的4.02M文件切分成（2.01M和2.01M）两个文件。<br />（2）切片过程：<br />判断虚拟存储的文件大小是否大于setMaxInputSplitSize值，大于等于则单独形成一个切片。如果不大于则跟下一个虚拟存储文件进行合并，共同形成一个切片。<br />测试举例：有4个小文件大小分别为1.7M、5.1M、3.4M以及6.8M这四个小文件，则虚拟存储之后形成6个文件块，大小分别为：<br />1.7M，（2.55M、2.55M），3.4M以及（3.4M、3.4M）<br />最终会形成3个切片，大小分别为：<br />（1.7+2.55）M，（2.55+3.4）M，（3.4+3.4）M
<a name="wseuz"></a>
### 2.3 MapReduce工作流程
![image.png](https://cdn.nlark.com/yuque/0/2022/png/22837646/1650809053134-dabb77db-f9e5-432d-9c9f-414ae3fb86d9.png#clientId=u5021c588-e66a-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=382&id=uc2b0fb1c&margin=%5Bobject%20Object%5D&name=image.png&originHeight=573&originWidth=1154&originalType=binary&ratio=1&rotation=0&showTitle=false&size=84678&status=done&style=stroke&taskId=udebd4cb4-9f2c-4368-bb28-218041ebfdb&title=&width=769.3333333333334)<br />![image.png](https://cdn.nlark.com/yuque/0/2022/png/22837646/1650809253525-fa6a01d0-526a-4b40-8ae6-f67c9333b11d.png#clientId=u5021c588-e66a-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=366&id=ud3442a76&margin=%5Bobject%20Object%5D&name=image.png&originHeight=549&originWidth=1134&originalType=binary&ratio=1&rotation=0&showTitle=false&size=59631&status=done&style=stroke&taskId=ua2f96951-ba56-410c-a07d-942c72db496&title=&width=756)<br />上面的流程是整个MapReduce最全工作流程，但是Shuffle过程只是从第7步开始到第16步结束，具体Shuffle过程详解，如下：<br />（1）MapTask收集我们的map()方法输出的kv对，放到内存缓冲区中<br />（2）从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件<br />（3）多个溢出文件会被合并成大的溢出文件<br />（4）在溢出过程及合并的过程中，都要调用Partitioner进行分区和针对key进行排序<br />（5）ReduceTask根据自己的分区号，去各个MapTask机器上取相应的结果分区数据<br />（6）ReduceTask会抓取到同一个分区的来自不同MapTask的结果文件，ReduceTask会将这些文件再进行合并（归并排序）<br />（7）合并成大文件后，Shuffle的过程也就结束了，后面进入ReduceTask的逻辑运算过程（从文件中取出一个一个的键值对Group，调用用户自定义的reduce()方法）<br />**注意：**
1. Shuffle中的缓冲区大小会影响到MapReduce程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快
1. 缓冲区的大小可以通过参数调整，参数：mapreduce.task.io.sort.mb默认100M
1. ReduceTask=0，表示没有reduce阶段，输出文件个数和Map个数一致
1. ReduceTask 默认值是1，所以输出文件个数为一个
1. 如果数据分布不均匀，就有可能在Reduce阶段产生数据倾斜
1. ReduceTask数量并不是任意设置，还要考虑业务逻辑需求，有些情况下，需要计算全局汇总结果，就只能有1个ReduceTask
1. 具体多少个ReduceTask，需要根据集群性能而定
1. 如果分区数不是1，但是ReduceTask为1，则不会执行分区过程。因为在MapTask源码中，执行分区的前提是先判断ReduceNum个数是否大于1。不大于1肯定不执行
<a name="efsfn"></a>
### 2.4 Shuffle机制
Map方法之后，Reduce方法之前的数据处理过程称之为Shuffle。<br />![image.png](https://cdn.nlark.com/yuque/0/2022/png/22837646/1650809792271-acb1005c-3338-4316-b8bb-86c1f12b5b5f.png#clientId=u5021c588-e66a-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=387&id=uc886027e&margin=%5Bobject%20Object%5D&name=image.png&originHeight=581&originWidth=1137&originalType=binary&ratio=1&rotation=0&showTitle=false&size=98674&status=done&style=stroke&taskId=ud697e6ff-1bf5-40f2-ab40-df5a8a0da4d&title=&width=758)
<a name="qPJd8"></a>
#### 2.4.1 Partition分区
1. 场景
将统计结果按条件输出到不同的分区中。默认的hash分区用户无法控制哪个key存储到哪个分区。
```java
@Public
@Stable
public class HashPartitioner<K, V> extends Partitioner<K, V> {
    public HashPartitioner() {
    }
    public int getPartition(K key, V value, int numReduceTasks) {
        return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
    }
}

1. 自定义Partition ```java // 根据手机号前三位输出到不同文件中 public class ProvincePartition extends Partitioner { @Override public int getPartition(FlowBean flowBean, Text text, int i) {

    String phone = text.toString().substring(0, 3);
    if ("136".equals(phone)) {
        return 0;
    } else if ("137".equals(phone)) {
        return 1;
    } else if ("138".equals(phone)) {
        return 2;
    } else if ("139".equals(phone)) {
        return 3;
    } else {
        return 4;
    }

   } }

// 在job中设置自定义分区，reduce个数，默认1 job.setPartitionerClass(ProvincePartition.class); job.setNumReduceTasks(5);

3. **总结**
- 如果ReduceTask的数据>getPartition的结果数，则会多产生几个空的输出文件；
- 如果1<ReduceTask的数量<getPartition的结果数，则有一部分分区数据无处安放，会报错；
- 如果ReduceTask的数量=1，则不管MapTask端输出多个个分区文件，最终结果都交给这一个ReduceTask，最终也就只会产生一个结果文件；
- 分区好必须从0开始，逐一累加。
<a name="npv0m"></a>
#### 2.4.2 WritableComparable排序
MapTask和ReduceTask都会对数据按照key进行排序。<br />Map阶段两次排序：对于MapTask，它会将处理的结果暂时放在环形缓冲区中，**①当环形缓冲区使用率达到一定阈值后，再对缓冲区中的数据进行一次快速排序，**并将这些有序数据溢写到磁盘上，而当数据处理完毕后，它会**②对磁盘上所有文件进行归并排序。**<br />Reduce阶段一次排序：对于ReduceTask，它从每个MapTask上远程拷贝相应的数据文件，如果文件大小超过一定阈值，则溢写到磁盘上，否则存储在内存中。如果磁盘上文件数目达到一定阈值，则进行一次归并排序以生成一个更大文件；如果内存中文件大小或者数目超过一定阈值，则进行一次合并后将数据溢写到磁盘上。当所有数据拷贝完毕后，**①ReduceTask统一对内存和磁盘上的所有数据进行一次归并排序**。
- 自定义排序
```java
public class FlowBean implements WritableComparable<FlowBean> {
    ...
    @Override
    public int compareTo(FlowBean o) {
        // 按总流量倒序排
        if (this.sumFlow > o.getSumFlow()) {
            return -1;
        } else if (this.sumFlow < o.getSumFlow()) {
            return 1;
        } else {
            // 按上行流量正序排
            if (this.upload > o.getUpload()) {
                return 1;
            } else if (this.upload < o.getUpload()) {
                return -1;
            } else {
                return 0;
            }
        }
    }
}

2.4.3 Combiner合并

1. Combiner是MR程序中Mapper和Reducer之外的一种组件
2. Combiner组件的父类就是Reducer
3. Combiner和Redicer的区别在于运行的位置：
   1. Combiner是在每一个MapTask所在的节点运行，是一个局部合并，例如（a,1）,(a,1)会合并为（a,2）之后再传输到reducer端
   2. Reducer是接收全局所有的Mapper的输出结果
4. Combiner的意义就是对每一个MapTask的输出进行局部汇总，以减小网路传输量

5. Combiner能够应用的前提是不影响最终的业务逻辑，而且，Combiner的输出kv应该跟Reducer的输入kv类型一致；不适用于算平均值等场景。

   public class WordCountCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable intWritable = new IntWritable();
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable value : values) {
            sum += value.get();
        }
        intWritable.set(sum);
        context.write(key, intWritable);
    }
   }

   2.5 OutputFormat数据输出

   OutputFormat是MapReduce输出的基类，所有实现MapReduce输出都实现了OutputFormat接口。默认输出格式TextOutputFormat。

   自定义OutputFormat步骤

   1. 自定义一个类继承FileOutputFormat
   2. 改写RecordWriter，具体改写输出数据的方法writer()。

2.6 MapReduce内核源码解析

2.6.1 MapTask工作机制

1. Read阶段：MapTask通过InputFormat获得的RecordReader，从输入InputSplit中解析出一个个key/value。
2. Map阶段：该节点主要是将解析出的key/value交给用户编写map()函数处理，并产生一系列新的key/value。
3. Collect收集阶段：在用户编写map()函数中，当数据处理完成后，一般会调用OutputCollector.collect()输出结果。在该函数内部，它会将生成的key/value分区（调用Partitioner），并写入一个环形内存缓冲区中。
4. Spill阶段：即“溢写”，当环形缓冲区满后，MapReduce会将数据写到本地磁盘上，生成一个临时文件。需要注意的是，将数据写入本地磁盘之前，先要对数据进行一次本地排序，并在必要时对数据进行合并、压缩等操作。

   如果 Job 设置过 Combiner, 那么现在就是使用 Combiner 的时候了. 将有相同 Key 的 Key/Value 对的 Value 合并在起来, 减少溢写到磁盘的数据量. Combiner 会优化 MapReduce 的中间结果, 所以它在整个模型中会多次使用 。 那哪些场景才能使用 Combiner 呢? 从这里分析, Combiner 的输出是 Reducer 的输入, Combiner 绝不能改变最终的计算结果. Combiner 只应该用于那种 Reduce 的输入 Key/Value 与输出 Key/Value 类型完全一致, 且不影响最终结果的场景. 比如累加, 最大值等. Combiner 的使用一定得慎重, 如果用好, 它对 Job 执行效率有帮助, 反之会影响 Reducer 的最终结果。

溢写阶段详情：
步骤1：利用快速排序算法对缓存区内的数据进行排序，排序方式是，先按照分区编号Partition进行排序，然后按照key进行排序。这样，经过排序后，数据以分区为单位聚集在一起，且同一分区内所有数据按照key有序。
步骤2：按照分区编号由小到大依次将每个分区中的数据写入任务工作目录下的临时文件output/spillN.out（N表示当前溢写次数）中。如果用户设置了Combiner，则写入文件之前，对每个分区中的数据进行一次聚集操作。
步骤3：将分区数据的元信息写到内存索引数据结构SpillRecord中，其中每个分区的元信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过1MB，则将内存索引写到文件output/spillN.out.index中。

1. Merge阶段：当所有数据处理完成后，MapTask对所有临时文件进行一次合并，以确保最终只会生成一个数据文件。

   当所有数据处理完后，MapTask会将所有临时文件合并成一个大文件，并保存到文件output/file.out中，同时生成相应的索引文件output/file.out.index。
   在进行文件合并过程中，MapTask以分区为单位进行合并。对于某个分区，它将采用多轮递归合并的方式。每轮合并mapreduce.task.io.sort.factor（默认10）个文件，并将产生的文件重新加入待合并列表中，对文件排序后，重复以上过程，直到最终得到一个大文件。
   让每个MapTask最终只生成一个数据文件，可避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。
   MapTask基础配置：

2.6.2 ReduceTask工作机制

1. Copy阶段：ReduceTask从各个MapTask上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。

2. Sort阶段：在远程拷贝数据的同时，ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。按照MapReduce语义，用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。

   merge过程： Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活。merge有三种形式：①内存到内存；②内存到磁盘；③磁盘到磁盘。默认情况下第一种形式不启用。当内存中的数据量到达一定阈值，就启动内存到磁盘的merge。与map 端类似，这也是溢写的过程，这个过程中如果你设置有Combiner，也是会启用的，然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行，直到没有map端的数据时才结束，然后启动第三种磁盘到磁盘的merge方式生成最终的文件。

3. Reduce阶段：reduce()函数将计算结果写到HDFS上。

   2.7 MapReduce开发总结

4. 输入数据接口：InputFormat

（1）默认使用的实现类是：TextInputFormat
（2）TextInputFormat的功能逻辑是：一次读一行文本，然后将该行的起始偏移量作为key，行内容作为value返回。
（3）CombineTextInputFormat可以把多个小文件合并成一个切片处理，提高处理效率。

1. 逻辑处理接口：Mapper

用户根据业务需求实现其中三个方法：map() setup() cleanup ()

1. Partitioner分区

（1）有默认实现 HashPartitioner，逻辑是根据key的哈希值和numReduces来返回一个分区号；key.hashCode()&Integer.MAXVALUE % numReduces
（2）如果业务上有特别的需求，可以自定义分区。

1. Comparable排序

（1）当我们用自定义的对象作为key来输出时，就必须要实现WritableComparable接口，重写其中的compareTo()方法。
（2）部分排序：对最终输出的每一个文件进行内部排序。
（3）全排序：对所有数据进行排序，通常只有一个Reduce。
（4）二次排序：排序的条件有两个。

1. Combiner合并

Combiner合并可以提高程序执行效率，减少IO传输。但是使用时必须不能影响原有的业务处理结果。

1. 逻辑处理接口：Reducer

用户根据业务需求实现其中三个方法：reduce() setup() cleanup ()

1. 输出数据接口：OutputFormat

（1）默认实现类是TextOutputFormat，功能逻辑是：将每一个KV对，向目标文本文件输出一行。
（2）用户还可以自定义OutputFormat。

3. Hadoop数据压缩

3.1 概述

1）压缩的好处和坏处
压缩的优点：以减少磁盘IO、减少磁盘存储空间。
压缩的缺点：增加CPU开销。
2）压缩原则
（1）运算密集型的Job，少用压缩
（2）IO密集型的Job，多用压缩

3.2 MR支持的压缩编码

1）压缩算法对比介绍

2）压缩性能的比较

3.3 压缩方式选择

压缩方式选择时重点考虑：压缩/解压缩速度、压缩率（压缩后存储大小）、压缩后是否可以支持切片。

3.3.1 Gzip压缩

优点：压缩率比较高；
缺点：不支持Split；压缩/解压速度一般；

3.3.2 Bzip2压缩

优点：压缩率高；支持Split；
缺点：压缩/解压速度慢。

3.3.3 Lzo压缩

优点：压缩/解压速度比较快；支持Split；
缺点：压缩率一般；想支持切片需要额外创建索引。

3.3.4 Snappy压缩

优点：压缩和解压缩速度快；
缺点：不支持Split；压缩率一般；

3.4 压缩位置选择

压缩可以在MapReduce作用的任意阶段启用。

3.5 压缩参数配置

1. 为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器。 | 压缩格式 | 对应的编码/解码器 | | —- | —- | | DEFLATE | org.apache.hadoop.io.compress.DefaultCodec | | gzip | org.apache.hadoop.io.compress.GzipCodec | | bzip2 | org.apache.hadoop.io.compress.BZip2Codec | | LZO | com.hadoop.compression.lzo.LzopCodec | | Snappy | org.apache.hadoop.io.compress.SnappyCodec |

2. 要在Hadoop中启用压缩，可以配置如下参数 | 参数 | 默认值 | 阶段 | 建议 | | —- | —- | —- | —- | | io.compression.codecs
   （在core-site.xml中配置） | 无，这个需要在命令行输入hadoop checknative查看 | 输入压缩 | Hadoop使用文件扩展名判断是否支持某种编解码器 | | mapreduce.map.output.compress（在mapred-site.xml中配置） | false | mapper输出 | 这个参数设为true启用压缩 | | mapreduce.map.output.compress.codec（在mapred-site.xml中配置） | org.apache.hadoop.io.compress.DefaultCodec | mapper输出 | 企业多使用LZO或Snappy编解码器在此阶段压缩数据 | | mapreduce.output.fileoutputformat.compress（在mapred-site.xml中配置） | false | reducer输出 | 这个参数设为true启用压缩 | | mapreduce.output.fileoutputformat.compress.codec（在mapred-site.xml中配置） | org.apache.hadoop.io.compress.DefaultCodec | reducer输出 | 使用标准工具或者编解码器，如gzip和bzip2 |