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

一、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支持的压缩编码

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

为了支持多种压缩/解压缩算法，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

压缩性能的比较:

压缩算法	原始文件大小	压缩文件大小	压缩速度	解压速度
gzip	8.3GB	1.8GB	2.9GB	58MB/s
bzip2	8.3GB	1.1GB	2.4MB/s	9.5MB/s
LZO	8.3GB	2.9GB	49.3MB/s	74.6MB/s

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 压缩位置选择

1.6 压缩参数配置

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

参数	默认值	阶段	建议
io.compression.codecs（在core-site.xml中配置）	org.apache.hadoop.io.compress.DefaultCodec, org.apache.hadoop.io.compress.GzipCodec, org.apache.hadoop.io.compress.BZip2Codec	输入压缩	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输出	这个参数设为true启用压缩
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
mapreduce.output.fileoutputformat.compress.type（在mapred-site.xml中配置）	RECORD	reducer输出	SequenceFile输出使用的压缩类型：NONE和BLOCK

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概述

2.1 Yarn的工作机制

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）
   
2. 容量调度器（Capacity Scheduler）
   
3. 公平调度器（Fair Scheduler）
   

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）：

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

YARN配置（yarn-default.xml）:

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

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

配置参数	参数说明
mapreduce.task.io.sort.mb	Shuffle的环形缓冲区大小，默认100m
mapreduce.map.sort.spill.percent	环形缓冲区溢出的阈值，默认80%

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

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

3.3 HDFS小文件优化方法

HDFS上每个文件都要在NameNode上建立一个索引，这个索引的大小约为150byte，这样当小文件比较多的时候，就会产生很多的索引文件，一方面会大量占用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之间。