Hadoop：数据压缩、Yarn、企业优化

亦凉 2023-02-14 13:27 23阅读 0赞

### 文章目录 ###

*  一、Hadoop数据压缩
 *   *  1.1 概述
     *  1.2 压缩策略和原则
     *  1.3 MR支持的压缩编码
     *  1.4 压缩方式选择
     *  1.5 压缩位置选择
     *  1.6 压缩参数配置
     *  1.7 压缩实操案例
 *  二、Yarn资源调度
 *   *  2.1 Yarn的工作机制
     *  2.2 资源调度器
     *  2.3 任务的推测执行
 *  三、企业优化
 *   *  3.1 MapReduce 跑的慢的原因
     *  3.2 MapReduce优化方法
     *   *  ①数据输入
         *  ②Map阶段
         *  ③Reduce阶段
         *  ④IO传输
         *  ⑤数据倾斜问题
         *  ⑥常用的调优参数
     *  3.3 HDFS小文件优化方法

# 一、Hadoop数据压缩 #

## 1.1 概述 ##

压缩技术能够有效减少底层存储系统（`HDFS`）读写字节数。压缩提供了网络带宽和磁盘空间的效率。在运行`MR`程序时，`IO`操作、网络数据传输、`Shuffle`和`Merge`要花大量的时间，尤其是数据规模很大和工作负载密集的情况下，因此，使用数据压缩显得非常重要。

鉴于磁盘`IO`和网络带宽是`Hadoop`的宝贵资源，数据压缩对于节省资源、最小化磁盘`IO`和网络传输资源非常有帮助。可以在任意`MapReduce`阶段启动压缩。不过，尽管压缩和解压操作的`CPU`开销不高，其性能的提升和资源的节省并非没有代价。

## 1.2 压缩策略和原则 ##

压缩是提供`Hadoop`运行效率的一种优化策略。通过对`Mapper`、`Reducer`运行过程中的数据进行压缩，以减少磁盘`IO`，提供`MR`程序运行速度。

**注意：** 采用压缩技术减少了磁盘`IO`,但同时增加了`CPU`运算负担。所以，压缩特性运用得当能提高性能，但运用不当也可能降低性能。

**压缩基本原则：**

1.  运算密集的`job`，少用压缩
2.  `IO`密集的`job`，多用压缩

## 1.3 MR支持的压缩编码 ##

<table> 
 <thead> 
  <tr> 
   <th>压缩格式</th> 
   <th>hadoop自带？</th> 
   <th>算法</th> 
   <th>文件扩展名</th> 
   <th>是否可切分</th> 
   <th>换成压缩格式后，原来的程序是否需要修改</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>DEFLATE</td> 
   <td>是，直接使用</td> 
   <td>DEFLATE</td> 
   <td>.deflate</td> 
   <td>否</td> 
   <td>和文本处理一样，不需要修改</td> 
  </tr> 
  <tr> 
   <td>Gzip</td> 
   <td>是，直接使用</td> 
   <td>DEFLATE</td> 
   <td>.gz</td> 
   <td>否</td> 
   <td>和文本处理一样，不需要修改</td> 
  </tr> 
  <tr> 
   <td>bzip2</td> 
   <td>是，直接使用</td> 
   <td>bzip2</td> 
   <td>.bz2</td> 
   <td>是</td> 
   <td>和文本处理一样，不需要修改</td> 
  </tr> 
  <tr> 
   <td>LZO</td> 
   <td>否，需要安装</td> 
   <td>LZO</td> 
   <td>.lzo</td> 
   <td>是</td> 
   <td>和文本处理一样，不需要修改</td> 
  </tr> 
  <tr> 
   <td>Snappy</td> 
   <td>否，需要安装</td> 
   <td>Snappy</td> 
   <td>.snappy</td> 
   <td>否</td> 
   <td>和文本处理一样，不需要修改</td> 
  </tr> 
 </tbody> 
</table>

为了支持多种压缩/解压缩算法，`Hadoop`引入了编码/解码器，如下表所示:

<table> 
 <thead> 
  <tr> 
   <th>压缩格式</th> 
   <th>对应的编码/解码器</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>DEFLATE</td> 
   <td>org.apache.hadoop.io.compress.DefaultCodec</td> 
  </tr> 
  <tr> 
   <td>gzip</td> 
   <td>org.apache.hadoop.io.compress.GzipCodec</td> 
  </tr> 
  <tr> 
   <td>bzip2</td> 
   <td>org.apache.hadoop.io.compress.BZip2Codec</td> 
  </tr> 
  <tr> 
   <td>LZO</td> 
   <td>com.hadoop.compression.lzo.LzopCodec</td> 
  </tr> 
  <tr> 
   <td>Snappy</td> 
   <td>org.apache.hadoop.io.compress.SnappyCodec</td> 
  </tr> 
 </tbody> 
</table>

压缩性能的比较:

<table> 
 <thead> 
  <tr> 
   <th>压缩算法</th> 
   <th>原始文件大小</th> 
   <th>压缩文件大小</th> 
   <th>压缩速度</th> 
   <th>解压速度</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>gzip</td> 
   <td>8.3GB</td> 
   <td>1.8GB</td> 
   <td>2.9GB</td> 
   <td>58MB/s</td> 
  </tr> 
  <tr> 
   <td>bzip2</td> 
   <td>8.3GB</td> 
   <td>1.1GB</td> 
   <td>2.4MB/s</td> 
   <td>9.5MB/s</td> 
  </tr> 
  <tr> 
   <td>LZO</td> 
   <td>8.3GB</td> 
   <td>2.9GB</td> 
   <td>49.3MB/s</td> 
   <td>74.6MB/s</td> 
  </tr> 
 </tbody> 
</table>

## 1.4 压缩方式选择 ##

**①Gzip压缩**

优点 :压缩率比较高，而且压缩/解压速度也比较快; `Hadoop`本身支持，在应用中处理`Gzip`格式的文件就和直接处理文本一样;大部分Linux系统都自带`Gzip`命令，使用方便

缺点：不支持`split`

应用场景：当每个文件压缩后130M以内的(一个块大小以内的)，都可以考虑用`Gzip`压缩格式，例如将一天或者一个小时的日子压缩成一个`Gzip`文件

**②Bzip2压缩**

优点：支持`Split`，具有很高的压缩率，比`Gzip`压缩率都高; `Hadoop`本身自带，使用方便

缺点：压缩/解压速度慢

应用场景：适合对速度要求不高，但需要较高的压缩率的时候;或者输出之后的数据比较大，处理之后的数据需要压缩存档减少磁盘空间并且以后数据用得比较少的情况;或者对单个很大的文本文件想压缩减少存储空间，同时又需要支持`Split`，而且兼容之前的应用程序的情况

**③Lzo压缩**

优点：压缩/解压速度也比较快，合理的压缩率；支持`Split`， 是`Hadoop`中最流行的压缩格式；可以在`Linux`系统下安装`lzop`命令，使用方便。

缺点：压缩率比`Gzip`要低一些； `Hadoop`本身不支持，需要安装；在应用中对`Lzo`格式的文件需要做一些特殊处理(为了支持`Split`需要建索引，还需要指定`InputFormat`为`Lzo`格式)。

应用场景：一个很大的文本文件，压缩之后还大于200M以上的可以考虑，而且单个文件越大，`Lzo`优点越越明显。

**④Snappy压缩**

优点：高速压缩速度和合理的压缩率

缺点：不支持`Split`；压缩率比`Gzip`要低； `Hadoop`本身不支持， 需要安装

应用场景：当`MapReduce`作业的`Map`输出的数据比较大的时候，作为`Map`到`Reduce`的中间数据的压缩格式；或者作为一个`MapReduce`作业的输出和另外一个`MapReduce`作业的输入

## 1.5 压缩位置选择 ##

![在这里插入图片描述][20200603150021560.png]

## 1.6 压缩参数配置 ##

要在`Hadoop`中启用压缩，可以配置如下参数：

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>默认值</th> 
   <th>阶段</th> 
   <th>建议</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>io.compression.codecs（在core-site.xml中配置）</td> 
   <td>org.apache.hadoop.io.compress.DefaultCodec,<br>org.apache.hadoop.io.compress.GzipCodec,<br>org.apache.hadoop.io.compress.BZip2Codec</td> 
   <td>输入压缩</td> 
   <td>Hadoop使用文件扩展名判断是否支持某种编解码器</td> 
  </tr> 
  <tr> 
   <td>mapreduce.map.output.compress（在mapred-site.xml中配置）</td> 
   <td>false</td> 
   <td>mapper输出</td> 
   <td>这个参数设为true启用压缩</td> 
  </tr> 
  <tr> 
   <td>mapreduce.map.output.compress.codec（在mapred-site.xml中配置）</td> 
   <td>org.apache.hadoop.io.compress.DefaultCodec</td> 
   <td>mapper输出</td> 
   <td>这个参数设为true启用压缩</td> 
  </tr> 
  <tr> 
   <td>mapreduce.output.fileoutputformat.compress（在mapred-site.xml中配置</td> 
   <td>false</td> 
   <td>reducer输出</td> 
   <td>这个参数设为true启用压缩</td> 
  </tr> 
  <tr> 
   <td>mapreduce.output.fileoutputformat.compress.codec（在mapred-site.xml中配置）</td> 
   <td>org.apache.hadoop.io.compress. DefaultCodec</td> 
   <td>reducer输出</td> 
   <td>使用标准工具或者编解码器，如gzip和bzip2</td> 
  </tr> 
  <tr> 
   <td>mapreduce.output.fileoutputformat.compress.type（在mapred-site.xml中配置）</td> 
   <td>RECORD</td> 
   <td>reducer输出</td> 
   <td>SequenceFile输出使用的压缩类型：NONE和BLOCK</td> 
  </tr> 
 </tbody> 
</table>

## 1.7 压缩实操案例 ##

**在驱动类`Driver`进行配置设置（具体参数见上）：**

public class WordCountDriver { 
    
    	public static void main(String[] args) throws Exception { 
    
    		Configuration configuration = new Configuration();
    
    		// 开启map端输出压缩
    		configuration.setBoolean("mapreduce.map.output.compress", true);
    		// 设置map端输出压缩方式
    		configuration.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, 
    			CompressionCodec.class);
    			
    		Job job = Job.getInstance(configuration);
    		......
    	}
    }

# 二、Yarn资源调度 #

`Yarn`架构见：[Hadoop概述][Hadoop]

## 2.1 Yarn的工作机制 ##

![在这里插入图片描述][20200603194356533.png]

1.  `MR`程序提交到客户端所在的节点
2.  `YarnRunner`向`ResourceManager`申请一个`Application`，`ResourceManager`将应用程序的资源路径返回给`YarnRunner`，然后将运行所需资源提交到`HDFS`上
3.  程序提交完毕后，申请运行`mrAppMaster`
4.  `ResourceManager`将用户的请求初始化成一个`Task`
5.  其中一个`NodeManager`领取到`Task`任务，该`NodeManager`创建容器`Container`并产生`MRAppmaster`
6.  `Container`从`HDFS`上拷贝资源到本地，`MRAppmaster`向`ResourceManager`申请运行`MapTask`资源
7.  `RM`将运行`MapTask`任务分配给另外两个`NodeManager`，另两个`NodeManager`分别领取任务并创建容器
8.  `MR`向两个接收到任务的`NodeManager`发送程序启动脚本，这两个`NodeManager`分别启动`MapTask`，`MapTask`对数据分区排序
9.  `MrAppMaster`等待所有`MapTask`运行完毕后，向`RM`申请容器，运行`ReduceTask`
10. `ReduceTask`向`MapTask`获取相应分区的数据
11. 程序运行完毕后，`MR`会向`RM`申请注销自己

## 2.2 资源调度器 ##

目前，`Hadoop`作业调度器主要有三种：`FIFO`、`Capacity Scheduler`和`Fair Scheduler`。`Hadoop2.7.2`默认的资源调度器是`Capacity Scheduler`。

具体设置详见：`yarn-default.xml`文件：

<property>
        <description>The class to use as the resource scheduler.</description>
        <name>yarn.resourcemanager.scheduler.class</name>
    	<value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler</value>
    </property>

1.  **先进先出调度器（FIFO）**  
    ![在这里插入图片描述][20200603201508306.png]
2.  **容量调度器（Capacity Scheduler）**  
    ![在这里插入图片描述][20200603201542575.png]
3.  **公平调度器（Fair Scheduler）**  
    ![在这里插入图片描述][20200603201558518.png]

## 2.3 任务的推测执行 ##

一个作业由若干个`Map`任务和`Reduce`任务构成。因硬件老化、软件`Bug`等，某些任务可能运行非常慢。

思考：系统中有99%的`Map`任务都完成了，只有少数几个`Map`老是进度很慢，完不成，怎么办？

**推测执行机制：** 发现拖后腿的任务，比如某个任务运行速度远慢于任务平均速度。为拖后腿任务启动一个备份任务，同时运行。谁先运行完，则采用谁的结果。

**执行推测任务的前提条件：**

1.  每个`Task`只能有一个备份任务
2.  当前`Job`已完成的`Task`必须不小于5%
3.  开启推测执行参数设置。`mapred-site.xml`文件中默认是打开的

<property>
      	<name>mapreduce.map.speculative</name>
      	<value>true</value>
      	<description>If true, then multiple instances of some map tasks may be executed in parallel.</description>
    </property>
    
    <property>
      	<name>mapreduce.reduce.speculative</name>
      	<value>true</value>
      	<description>If true, then multiple instances of some reduce tasks may be executed in parallel.</description>
    </property>

**不能启用推测执行机制情况：**

1.  任务间存在严重的负载倾斜
2.  特殊任务，比如任务向数据库中写数据

# 三、企业优化 #

## 3.1 MapReduce 跑的慢的原因 ##

**①计算机性能**‘

CPU、内存、磁盘健康、网络

**②I/O操作优化**

1.  数据倾斜
2.  `MapReduce`数设置不合理
3.  `Map`运行时间太长，导致`Reduce`等待过久
4.  小文件过多
5.  大量的不可分块的超大文件
6.  `Spil`次数过多
7.  `Merge`次数过多等

## 3.2 MapReduce优化方法 ##

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

### ①数据输入 ###

合并小文件:在执行`MR`任务前将小文件进行合并，大量的小文件会产生大量的`Map`任务，增大`Map`任务装载次数，而任务的装载比较耗时，从而导致`MR`运行较慢。

采用`Combine TextInputFormat`来作为输入，解决输入端大量小文件场景

### ②Map阶段 ###

1.  减少溢写`Spill`次数：通过调整`io.sort.mb`及`sort.spill,percent`参数值，增大触发`Spill`的内存上限，减少`Spill`次数，从而减少磁盘`IO`
2.  减少合并`Merge`次数：通过调整`io.sort.factor`参数，增大`Merge`的文件数目，减少`Merge`的次数，从而缩短`MR`的处理时间
3.  在`Map`之后，不影响业务逻辑的前提下，先进行`Combiner`处理，减少`IO`

### ③Reduce阶段 ###

1.  合理设置`Map`和`Reduce`数：两个都不能设置太少，也不能设置太多。太少，会导致`Task`等待，延长处理时间；太多，会导致`Map`、 `Reduce`任务间竞争资源，造成处理超时等错误。
2.  设置`Map`、`Reduce`共存：调整`slowstart.completedmaps`参数，使`Map`运行到一定程度后，`Reduce`也开始运行，减少`Reduce`的等待时间。
3.  规避使用`Reduce`：因为`Reduce`在用于连接数据集的时候将会产生大量的网络消耗。
4.  合理设置`Reduce`端的`Buffer`：默认情况下，数据达到一个阈值的时候，`Buffer`中的数据就会写入磁盘，然后`Reduce`会从磁盘中获得所有的数据。也就是说，`Buffer`和`Reduce`是没有直接关联的，中间多次写磁盘`->`读磁盘的过程，既然有这个弊端，那么就可以通过参数来配置，使得`Buffer`中的一部分数据可以直接输送到`Reduce`，从而减少`IO`开销： `mapreduce.reduce input. buffer. percent`，默认为0.0。 当值大于0的时候，会保留指定比例的内存读`Buffer`中的数据直接拿给`Reduce`使用。这样一来，设置`Buffer`需要内存，读取数据需要内存，`Reduce`计算也要内存，所以要根据作业的运行情况进行调整。

### ④IO传输 ###

1.  采用数据压缩的方式，减少网络`IO`的的时间，安装`Snappy`和`LZO`压缩编码器。
2.  使用`SequenceFile`二进制文件。

### ⑤数据倾斜问题 ###

**数据倾斜现象：**

1.  数据频率倾斜：某一个区域的数据量要远远大于其他区域。
2.  数据大小倾斜：部分记录的大小远远大于平均值。

**减少数据倾斜的方法：**

1.  **抽样和范围分区**：可以通过对原始数据进行抽样得到的结果集来预设分区边界值。
2.  **自定义分区：** 基于输出键的背景知识进行自定义分区。例如，如果`Map`输出键的单词来源于一本书。且其中某几个专业词汇较多。那么就可以自定义分区将这这些专业词汇发送给固定的一部分`Reduce`实例， 而将其他的都发送给剩余的`Reduce`实例。
3.  **Combine**： 使用`Combine`可以大量地减小数据倾斜。在可能的情况下，`Combine`的目的就是聚合并精简数据。
4.  **采用Map Join，尽量避免Reduce Join**

### ⑥常用的调优参数 ###

**资源相关参数（配置在`mapred-default.xml`）：**

<table> 
 <thead> 
  <tr> 
   <th>配置参数</th> 
   <th>参数说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>mapreduce.map.memory.mb</td> 
   <td>一个MapTask可使用的资源上限（单位:MB），默认为1024。如果MapTask实际使用的资源量超过该值，则会被强制杀死</td> 
  </tr> 
  <tr> 
   <td>mapreduce.reduce.memory.mb</td> 
   <td>一个ReduceTask可使用的资源上限（单位:MB），默认为1024。如果ReduceTask实际使用的资源量超过该值，则会被强制杀死</td> 
  </tr> 
  <tr> 
   <td>mapreduce.map.cpu.vcores</td> 
   <td>每个MapTask可使用的最多cpu core数目，默认值: 1</td> 
  </tr> 
  <tr> 
   <td>mapreduce.reduce.cpu.vcores</td> 
   <td>每个ReduceTask可使用的最多cpu core数目，默认值: 1</td> 
  </tr> 
  <tr> 
   <td>mapreduce.reduce.shuffle.parallelcopies</td> 
   <td>每个Reduce去Map中取数据的并行数。默认值是5</td> 
  </tr> 
  <tr> 
   <td>mapreduce.reduce.shuffle.merge.percent</td> 
   <td>Buffer中的数据达到多少比例开始写入磁盘。默认值0.66</td> 
  </tr> 
  <tr> 
   <td>mapreduce.reduce.shuffle.input.buffer.percent</td> 
   <td>Buffer大小占Reduce可用内存的比例。默认值0.7</td> 
  </tr> 
  <tr> 
   <td>mapreduce.reduce.input.buffer.percent</td> 
   <td>指定多少比例的内存用来存放Buffer中的数据，默认值是0.0</td> 
  </tr> 
 </tbody> 
</table>

**YARN配置（`yarn-default.xml`）:**

<table> 
 <thead> 
  <tr> 
   <th>配置参数</th> 
   <th>配置参数</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>yarn.scheduler.minimum-allocation-mb</td> 
   <td>给应用程序Container分配的最小内存，默认值：1024</td> 
  </tr> 
  <tr> 
   <td>yarn.scheduler.maximum-allocation-mb</td> 
   <td>给应用程序Container分配的最大内存，默认值：8192</td> 
  </tr> 
  <tr> 
   <td>yarn.scheduler.maximum-allocation-mb</td> 
   <td>每个Container申请的最小CPU核数，默认值：1</td> 
  </tr> 
  <tr> 
   <td>yarn.scheduler.maximum-allocation-vcores</td> 
   <td>每个Container申请的最大CPU核数，默认值：32</td> 
  </tr> 
  <tr> 
   <td>yarn.nodemanager.resource.memory-mb</td> 
   <td>给Containers分配的最大物理内存，默认值：8192</td> 
  </tr> 
 </tbody> 
</table>

**Shuffle性能优化（`mapred-default.xml`）：**

<table> 
 <thead> 
  <tr> 
   <th>配置参数</th> 
   <th>参数说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>mapreduce.task.io.sort.mb</td> 
   <td>Shuffle的环形缓冲区大小，默认100m</td> 
  </tr> 
  <tr> 
   <td>mapreduce.map.sort.spill.percent</td> 
   <td>环形缓冲区溢出的阈值，默认80%</td> 
  </tr> 
 </tbody> 
</table>

**容错相关参数(`MapReduce`性能优化)：**

<table> 
 <thead> 
  <tr> 
   <th>配置参数</th> 
   <th>参数说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>mapreduce.map.maxattempts</td> 
   <td>每个MapTask最大重试次数，一旦重试参数超过该值，则认为MapTask运行失败，默认值：4</td> 
  </tr> 
  <tr> 
   <td>mapreduce.reduce.maxattempts</td> 
   <td>每个ReduceTask最大重试次数，一旦重试参数超过该值，则认为ReduceTask运行失败，默认值：4</td> 
  </tr> 
  <tr> 
   <td>mapreduce.task.timeout</td> 
   <td>Task超时时间，经常需要设置的一个参数，该参数表达的意思为：如果一个Task在一定时间内没有任何进入，即不会读取新的数据，也没有输出数据，则认为该Task处于Block状态，可能是卡住了，也许永远会卡住，为了防止因为用户程序永远Block住不退出，则强制设置了一个该超时时间（单位毫秒），默认是600000。如果你的程序对每条输入数据的处理时间过长（比如会访问数据库，通过网络拉取数据等），建议将该参数调大，该参数过小常出现的错误提示是“AttemptID:attempt_14267829456721_123456_m_000224_0 Timed out after 300 secsContainer killed by the ApplicationMaster.”。</td> 
  </tr> 
 </tbody> 
</table>

## 3.3 HDFS小文件优化方法 ##

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

1.  **Hadoop Archive**  
    是一个高效地将小文件放入`HDFS`块中的文件存档工具，它能够将多个小文件打包成-一个HAR文件，这样就减少了NameNode的内存使用。
2.  **Sequence File**  
    `Sequence File`由一系列的二进制`key/value`组成，如果`key`为文件名，`value`为文件内容，则可以将大批小文件合并成一个大文件。
3.  **CombineFilelnputFormat**  
    `CombineFileInputFormat`是一种新的`InputFormat`， 用于将多个文件合并成一个单独的`Split`，另外，它会考虑数据的存储位置。
4.  **开启JVM重用.**  
    对于大量小文件`Job`，可以开启`JVM`重用会减少45%运行时间。  
      
    `JVM`重用原理: 一个`Map`运行在一个`JVM`上，开启重用的话，该`Map`在`JVM`上运行完毕后，`JVM`继续运行其他`Map`。  
      
    具体设置: `mapreduce.job.jvm.numtasks`值在10-20之间。

[20200603150021560.png]: https://img-blog.csdnimg.cn/20200603150021560.png
[Hadoop]: https://hucheng.blog.csdn.net/article/details/102935014
[20200603194356533.png]: https://img-blog.csdnimg.cn/20200603194356533.png
[20200603201508306.png]: https://img-blog.csdnimg.cn/20200603201508306.png
[20200603201542575.png]: https://img-blog.csdnimg.cn/20200603201542575.png
[20200603201558518.png]: https://img-blog.csdnimg.cn/20200603201558518.png