[大数据]009 优化&新特性&HA

1、Hadoop数据压缩

• 输入压缩：（Hadoop使用文件扩展名判断是否支持某种编解码器，core-site.xml）
  org.apache.hadoop.io.compress.DefaultCodec
  org.apache.hadoop.io.compress.GzipCodec
  org.apache.hadoop.io.compress.BZip2Codec
  com.hadoop.compression.lzo.LzopCodec
  org.apache.hadoop.io.compress.SnappyCodec
• mapper输出：（企业多使用LZO或Snappy编解码器在此阶段压缩数据，mapred-site.xml）
  com.hadoop.compression.lzo.LzopCodec
  org.apache.hadoop.io.compress.SnappyCodec
• reducer输出：（使用标准工具或者编解码器，如gzip和bzip2，mapred-site.xml）
  org.apache.hadoop.io.compress.GzipCodec
  org.apache.hadoop.io.compress.BZip2Codec

PS：LZO格式是基于GPL许可的，不能通过Apache来分发许可，基于此，它的hadoop编码/解码器必须单独下载，Linux上安装编译lzo详解。lzop编码/解码器兼容干lzop工具，它其实就是LZO 格式，但额外还有头部，它正是我们想要的。还有一个纯LZO格式的编码/解码器LzoCodec，它使用.lzo_deflate作为扩展名(根据 DEFLATE类推，是没有头部的gzip格式）。

1.1、数据流的压缩和解压缩

//获取压缩编解码器codec
CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());
CompressionCodec codec = factory.getCodecByName(method);
//获取普通输出流,文件后面需要加上压缩后缀
FileOutputStream fos = new FileOutputStream(new File(filename + codec.getDefaultExtension()));
//获取压缩输出流,用压缩解码器对fos进行压缩
CompressionOutputStream cos = codec.createOutputStream(fos);

//获取压缩编解码器codec
CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());
CompressionCodec codec = factory.getCodec(new Path(filename));
//获取普通输入流
FileInputStream fis = new FileInputStream(new File(filename));
//获取压缩输出流,用压缩解码器对fis进行解压
CompressionInputStream cis = codec.createInputStream(fis);

1.2、Map、Reduce输出端采用压缩

Mapper和Reducer不变

// 开启map端输出压缩
conf.setBoolean("mapreduce.map.output.compress", true);
// 设置map端输出压缩方式
conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class,CompressionCodec.class);

// 设置reduce端输出压缩开启
FileOutputFormat.setCompressOutput(job, true);
// 设置压缩的方式
FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);

2、Hadoop企业优化

2.1、MapReduce程序效率的瓶颈

1）计算机性能：CPU、内存、硬盘、网络
2）I/O操作优化：数据倾斜、MapTask和ReduceTask数不合理、小文件、压缩文件不可切分、切片数过多、Merge数过多、Reduce时间过长

解决方案：
1）输入阶段：CombineTextInputFormat合并输入端大量的小文件
2）Map阶段：减少溢写次数、减少合并次数、加入Combine
mapred-default.xml

<!-- 增大触发Spill的内存上限-->
<property>
  <name>mapreduce.task.io.sort.mb</name>
  <value>100</value>
  <description>The total amount of buffer memory to use while sorting
  files, in megabytes.  By default, gives each merge stream 1MB, which
  should minimize seeks.</description>
</property>
<property>
  <name>mapreduce.map.sort.spill.percent</name>
  <value>0.80</value>
  <description>The soft limit in the serialization buffer. Once reached, a
  thread will begin to spill the contents to disk in the background. Note that
  collection will not block if this threshold is exceeded while a spill is
  already in progress, so spills may be larger than this threshold when it is
  set to less than .5</description>
</property>

<!--增大Merge的文件数目-->
<property>
  <name>mapreduce.task.io.sort.factor</name>
  <value>10</value>
  <description>The number of streams to merge at once while sorting
  files.  This determines the number of open file handles.</description>
</property>

3）Reduce阶段：合理设置MapTask和ReduceTask数（太少task会等待，太多task会竞争）、设置Map和Reduce共存（Map运行到一定程度后，开始运行Reduce）、减少Reduce（Reduce获取数据产生大量的网络消耗）
mapred-default.xml

<property>
  <name>mapreduce.job.reduce.slowstart.completedmaps</name>
  <value>0.05</value>
  <description>Fraction of the number of maps in the job which should be
  complete before reduces are scheduled for the job.
  </description>
</property>

<property>
  <name>mapreduce.reduce.input.buffer.percent</name>
  <value>0.0</value>
  <description>The percentage of memory- relative to the maximum heap size- to
  retain map outputs during the reduce. When the shuffle is concluded, any
  remaining map outputs in memory must consume less than this threshold before
  the reduce can begin.
  </description>
</property>

4）I/O阶段：使用Snappy和LZO压缩编码器、使用SequenceFile二进制文件（对hive二进制存储格式,即SequenceFile和RCFile的思考总结）
5）数据倾斜：抽样和范围分区（数据抽样预设分区）、自定义分区、Combiner精简数据、避免Reduce Join（尽量Map Join）

2.2、hadoop常用的调优参数

2.3、Hadoop小文件优化方法

补充：SequenceFile是由一系列的二进制k/v组成，如果为key为文件名，value为文件内容，可将大批小文件合并成一个大文件

3、Hadoop新特性

3.1、采用distcp命令实现两个Hadoop集群之间的递归数据复制

[atguigu@hadoop102 hadoop-3.1.3]$  bin/hadoop distcp hdfs://hadoop102:9820/user/atguigu/hello.txt hdfs://hadoop105:9820/user/atguigu/hello.txt

3.2、小文件存档

# 归档文件
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop archive -archiveName input.har -p  /user/atguigu/input   /user/atguigu/output
# 查看归档
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -ls /user/atguigu/output/input.har
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -ls har:///user/atguigu/output/input.har
# 解归档文件
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -cp har:/// user/atguigu/output/input.har/*    /user/atguigu

3.3、Hadoop Trash回收站使用指南

补充：通过程序删除的文件不会经过回收站，需要调用moveToTrash()才进入回收站

Trash trash = New Trash(conf);
trash.moveToTrash(path);

3.4、Hadoop3.x新特性

PS：纠删码Erasure Coding （分布式存储系统）

3.5、Hadoop HA 高可用

hadoop中的JournalNode
HDFS HA（高可用）机制
使用Quorum Journal Manager(QJM)的HDFS NameNode高可用配置
【HDFS篇11】HA高可用
HDFS-HA：YARN的JournalNode实现EditLog的共享，Zookeeper的ZKFailoverController实现自动故障切换
Hadoop3 HA部署之Yarn篇
Hadoop(3)—如何构建HDFS–HA,YARN—HA
Hadoop3.1.2 高可用安装Yarn (ResourceManager High Availability)
YARN-HA：ResourceManager中ActiveStandbyElector有与Zookeeper进行交互的功能，无需使用ZKFailoverController。

3.6、HDFS Federation架构设计

NameNode架构的局限性（单NameNode）：
1）Namespace的限制（NameNode所能存储的对象（文件+块）数目受到NameNode所在JVM的heap size的限制）；
2）隔离问题（HDFS上的一个程序运行整个HDFS上的所有程序）；
3）性能的瓶颈（HDFS文件系统的吞吐量受限于单个NameNode的吞吐量）；