hive性能调优

Posted 2021-04-29 Hadoop大数据之路

Hadoop为海量数据存储和计算提供了一种能力，在Hadoop的计算框架特性下会衍生如下问题：

1、数据量大不是问题，数据倾斜是个问题；

2、jobs数比较多的作业运行效率较低，原因是map reduce作业初始化的时间比较长；

3、count(distinct)在数据量大的情况下，效率极低，如果是多count(distinct)效率更低，因为count(distinct)是按group by字段分组，按distinct字段排序，一般这种分布方式是很倾斜的。

面对这些问题，有哪些有效的优化手段呢？

一、行/列式存储特点

行式存储

优点：

相关的数据是保存在一起的，比较符合面向对象的思维，因为一行数据就是一条记录；这种存储格式比较方便进行INSERT/UPDATE操作。

缺点：

如果查询涉及某几列，它会把整行数据读取出来，不能跳过不必要的列读取，如果数据量比较大就比较影响性能；如果每一行中，列的数据类型不一致，导致不容易获得一个极高的压缩比，导致空间利用率不高；不是所有的列都适合作为索引。

列式存储

优点：

查询数据时，只有涉及的列才会被查询，不会把所有列查询出来，即可以跳过不必要的列查询；高效的压缩率，节省存储空间和计算内存CPU；任何列都可以作为索引。

缺点：

INSERT/UPDATE不方便；不适合扫描小量的数据。

二、hive文件存储格式

1、textfile

默认存储格式，存储方式：行存储。特点：磁盘开销大，数据解释开销大，压缩的text文件hive无法进行合并和拆分。

2、sequencefile

二进制文件，以<key, value>的形式序列化到文件中，存储方式：行存储。特点：可分割，可压缩，与hadoop的mapfile是互相兼容的。

3、rcfile

存储方式：数据按行分块，每块按照列存储。特点：压缩快，快速列存取。

4、orc

存储方式：数据按行分块，每块按照列存储。特点：压缩快，快速列存取，效率比rcfile高，是rcfile的改良版本。

5、自定义格式

可以通过实现InputFormat和OutputFormat来自定义输入输出格式。

三、hadoop四种压缩格式

1、gzip压缩

优点：压缩率比较高，而且压缩/解压速度较快，hadoop本身支持，在应用中处理gzip格式的文件和直接处理文本一样。

缺点：不支持split。

应用场景：当每个文件压缩之后在1个块大小内，都可以考虑用gzip压缩格式。

2、lzo压缩

优点：压缩/解压速度较快，合理的压缩率，支持split，是hadoop中最流行的压缩格式，支持hadoop native库，可以在linux系统下安装lzop命令，使用方便。

缺点：压缩率比gzip要低一些，hadoop本身不支持，需要安装，在应用中对lzo格式的文件需要做一些特殊处理。

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

3、snappy压缩

优点：高速压缩速度和合理的压缩率，支持hadoop native库。

缺点：不支持split，压缩率比gzip要低，hadoop本身不支持，需要安装，linux系统下没有对应的命令。

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

4、bzip2压缩

优点：支持split，具有很高的压缩率，比gzip压缩率都高，hadoop本身支持，但不支持native，在linux系统下自带bzip2命令，使用方便。

缺点：压缩/解压速度慢，不支持native。

应用场景：适合对速度要求不高，但需要较高的压缩率的情况。

在hive中对中间数据或最终数据做压缩，是提高数据吞吐量和性能的一种手段。对数据做压缩，可以大量减少磁盘的存储空间，压缩后的文件在磁盘间传输和I/O也会大大减少，当然压缩和解压也会带来额外的CPU开销，但是却可以节省更多的I/O和使用更少的内存。数据压缩对I/O密集型作业带来大大的性能提升，如果作业是CPU密集型则会降低性能。

四、调优策略

1、JVM重用

JVM重用不是指同一Job的两个或两个以上的Task同时运行于同一JVM上，而是 N个Task按顺序在同一个Jvm上运行，即省去了Jvm关闭和再重启的时间。

N值可以在Hadoop的mapre-site.xml文件mapreduce.job.jvm.numtasks(默认1)属性进行设置，也可以通过set mapred.job.reuse.jvm.num.tasks=10设置；

2、并行执行

同时执行hive的多个阶段。hive在执行过程，将一个查询转化成一个或者多个阶段，而这些阶段可能并非完全相互依赖的，也就是说可以并行执行的，这样可能使得整个job的执行时间缩短hive执行开启：set hive.exec.parallel=true；

3、调整reduce数

参数1：hive.exec.reducers.bytes.per.reducer=256000000(每个reduce任务处理的数据量)；参数2：hive.exec.reducers.max=1009(每个任务最大的reduce数目)；计算公式：reducer个数=min(参数2,总输入数据量/参数1)。set mapred.reduce.tasks=N(每个任务默认的reduce数目)，典型为0.99*reduce槽数，hive默认为-1，即自动确定reduce数目。reduce的个数设置其实对执行效率有很大的影响：1、如果reduce太少，如果数据量很大，会导致这个reduce异常的慢，从而导致这个任务不能结束，也有可能会OOM。2、如果reduce太多，产生的小文件太多，合并起来代价太高，namenode的内存占用也会增大。

4、调整map数

（1）默认map个数

如果不进行任何设置，默认的map个数是和blcok_size相关的。default_num = total_size / block_size

（2）期望大小

可以通过参数mapred.map.tasks来设置程序员期望的map个数，但是这个个数只有在大于default_num的时候，才会生效。

goal_num = mapred.map.tasks

（3）设置处理的文件大小

可以通过mapred.min.split.size 设置每个task处理的文件大小，但是这个大小只有在大于block_size的时候才会生效。

split_size = max(mapred.min.split.size, block_size)

split_num = total_size / split_size

（4）计算的map个数

compute_map_num = min(split_num,  max(default_num, goal_num))

除了这些配置以外，mapreduce还要遵循一些原则。 mapreduce的每一个map处理的数据是不能跨越文件的，也就是说min_map_num >= input_file_num。 所以，最终的map个数应该为：

final_map_num = max(compute_map_num, input_file_num)

结论：

1）如果想增加map个数，则设置mapred.map.tasks为一个较大的值。

2）如果想减小map个数，则设置mapred.min.split.size 为一个较大的值。

3）如果输入中有很多小文件，依然想减少map个数，则需要将小文件merger为大文件，然后使用准则2。

5、配置map输入合并

每个Map最大输入大小，决定合并后的文件数。

set mapred.max.split.size=256000000;

一个节点上split的至少的大小 ，决定了多个data node上的文件是否需要合并

set mapred.min.split.size.per.node=100000000;

一个交换机下split的至少的大小，决定了多个交换机上的文件是否需要合并

set mapred.min.split.size.per.rack=100000000;

执行Map前进行小文件合并

set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; 

6、配置hive输出结果合并

hive.merge.mapfiles 在map-only job后合并文件，默认true

hive.merge.mapredfiles 在map-reduce job后合并文件，默认false

hive.merge.size.per.task 合并后每个文件的大小，默认256000000

hive.merge.smallfiles.avgsize 平均文件大小，是决定是否执行合并操作的阈值，默认16000000

hive性能调优参数（仅供参考）：

#jvm重用

set mapred.job.reuse.jvm.num.tasks=10;

#并行执行

set hive.exec.parallel=true;

set hive.exec.parallel.thread.number=8;

#输出合并

set hive.merge.mapfiles=true;

set hive.merge.mapredfiles=true;

set hive.merge.size.per.task=256000000;

set hive.merge.smallfiles.avgsize=16000000;#输出文件的平均大小小于该值则启动一个独立的MR任务进行文件合并

#输入合并

set mapred.max.split.size=256000000; 

set mapred.min.split.size.per.node=256000000; 

set mapred.min.split.size.per.rack=256000000;

set hive.hadoop.supports.splittable.combineinputformat=true; 

set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; 

#输出压缩

set hive.exec.compress.output=true; 

set mapred.output.compress=true; 

set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec; 

set io.compression.codecs=org.apache.hadoop.io.compress.GzipCodec;