MapReduce性能优化秘籍

1. MapReduce跑的慢的原因

MapReduce 程序效率的瓶颈在于两点：

1. 计算机性能
   CPU、内存、磁盘、网络
2. I/O 操作

1. 数据倾斜
2. map 和 reduce 数设置不合理
3. map 运行时间太长，导致 reduce 等待过久
4. 小文件过多
5. 大量的不可分块的超大文件（例：通过 gzip 压缩后的文件）
6. spill（溢写）次数过多
7. merge（map 端合并或 reduce 端合并）次数过多

2. MapReduce优化方法

MapReduce 优化方法主要从六个方面考虑：数据输入、Map 阶段、Reduce 阶段、IO 传输、数据倾斜问题和常用的调优参数。

2.1. 数据输入

• 合并小文件：在执行 mr 任务前将小文件进行合并，大量的小文件会产生大量的 map 任务，增大 map 任务装载次数，而任务的装载比较耗时，从而导致 mr 运行较慢。
• 采用 CombineTextInputFormat 来作为输入，解决输入端大量小文件的场景。

2.2. Map阶段

• 减少溢写（spill）次数：通过调整 io.sort.mb增大环形缓冲区的大小；调整 sort.spill.percent减少溢写的频率，从而减少磁盘IO。
• 减少合并（merge）次数：通过调整 io.sort.factor参数，增大触发合并的文件数目，减少合并的次数，从而缩短 mr 处理时间。
• 在不影响业务逻辑的前提下，先使用 Combiner 在 map 端进行一次合并，减少 I/O。

2.3. Reduce阶段

• 合理设置 map 和 reduce 的个数：两个都不能设置太少，也不能设置太多。太少，会导致 task 等待，延长处理时间；太多，会导致 map、reduce 任务间竞争资源，造成处理超时等错误。
• 设置 map、reduce 共存：调整 slowstart.completedmaps参数，使 map 运行到一定程度后，reduce 也开始运行，减少 reduce 的等待时间。
• 规避使用 reduce：因为 reduce 阶段会通过网络去下载 map 运行的结果复制到 reduce 节点，造成大量的网络消耗。
• 合理设置 reduce 端的 buffer：默认情况下，从 map 端传输到 reduce 端放到 buffer 中的数据达到一个阙值的时候，buffer中的数据就会写入磁盘，然后 reduce 会从磁盘中获得所有的数据。也就是说，buffer 和 reduce 是没有直接关联的，中间多一个写磁盘读磁盘的过程，既然有这个弊端，那么就可以通过参数来配置，使得 buffer 中的一部分数据可以直接输送到 reduce，从而减少 IO 开销：mapred.job.reduce.input.buffer.percent，默认为 0.0 。当值大于 0 的时候，会保留指定比例的内存读 buffer 中的数据直接拿给 reduce 使用。这样一来，设置 buffer 需要内存，读取数据需要内存，reduce 计算也要内存，所以要根据作业的运行情况进行调整。

2.4. IO传输

• 启用数据压缩方式，减少网络 IO 的时间。常见的压缩方式 gzip，bzip2，LZO，Snappy。企业中多实用 LZO，Snappy，需要安装 LZO 和 Snappy 压缩编码器。
• 使用 Sequence File 二进制文件。

2.5. 数据倾斜问题

什么是数据倾斜？

• 数据倾斜现象：
  数据频率倾斜——某一个区域的数据量要远远大于其他区域。
  数据大小倾斜——部分记录大小远远大于平均值
• 如何收集倾斜数据
  在reduce 方法中加入记录 map 输出键的详细情况的功能。

public static final String MAX_VALUES = "skew.maxvalues"; 
private int maxValueThreshold; 
 
@Override
public void configure(JobConf job) { 
     maxValueThreshold = job.getInt(MAX_VALUES, 100); 
} 
@Override
public void reduce(Text key, Iterator<Text> values,
                     OutputCollector<Text, Text> output, 
                     Reporter reporter) throws IOException {
     int i = 0;
     while (values.hasNext()) {
         values.next();
         i++;
     }

     if (++i > maxValueThreshold) {
         log.info("Received " + i + " values for key " + key);
     }
}

如何解决数据倾斜？

方法1：抽样和范围分区
可以通过对原始数据进行抽样得到的结果集来预设分区边界值。

方法2：自定义分区
可以自定义分区。例如：如果 map 输出键的单词来源于一本书。且其中某几个专业词汇比较多。那么就可以自定义分区将这些专业词汇发送给固定的一部分 reduce 实例。而将其他的都发送给剩余的 reduce 实例。

方法3：使用 Combiner
使用 Combiner 可以大量减少数据倾斜。在可能的情况下，combine 的目的就是聚合并精简数据。

方法4：采用 Map Join，尽量避免 Reduce Join

2.6. 常用的调优参数

以下参数是在用户自己的 mr 应用程序中就可以生效（mapred-default.xml）

以下参数应该在 yarn 启动之前就配置在服务器的配置文件中才能生效（yarn-default.xml）

以下是 shuffle 性能优化的关键参数，应在 yarn 启动之前就配置好（mapred-default.xml）

容错相关参数：

3. HDFS小文件优化方法

3.1. HDFS小文件弊端

HDFS 上每个文件都要在 namenode 上建立一个索引，这个索引的大小约为 150byte，这个当小文件比较多的时候，就会产生很多的索引文件，一方面大量占用 namenode 的内存空间，另一方面就是索引文件过大使索引速度变慢。

3.2. 解决方案

1. Hadoop Archive：
   是一个高效地将小文件放入 HDFS 块中的文件存档工具，它能够将多个小文件打包成一个 HAR 文件，这样就减少了 namenode 的内存使用。
2. Sequence File：
   sequence file 由一系列二进制 key/value 组成，如果 key 为文件名，value 为文件内容，则可以将大批小文件合并成一个大文件。
3. CombineTextInputFormat：
   CombineTextInputFormat 是一种新的 inputformat，用于将多个文件从逻辑上合并成一个单独的切片，由一个 map task 处理。
4. 开启 JVM 重用
   对于大量小文件 Job，可以开启 JVM 重用会减少 45% 运行时间。
   JVM 重用理解：一个 map 运行一个 jvm，重用的话，在一个 jvm 上运行完毕后，jvm 继续运行其他 map。
   具体设置：mapreduce.job.jvm.numtasks值在 10 ~ 20 之间。