Hive优化

一、join优化

1. 使用相同的连接键
当对3个或者更多个表进行join连接时，如果每个on子句都使用相同的连接键的话，那么只会产生一个MapReduce job。

2. 尽量尽早地过滤数据
减少每个阶段的数据量,对于分区表要加分区，同时只选择需要使用到的字段。

3. 尽量原子化操作
尽量避免一个SQL包含复杂逻辑，可以使用中间表来完成复杂的逻辑。

二、小文件优化

1.小文件过多产生的影响
首先对底层存储HDFS来说，HDFS本身就不适合存储大量小文件，小文件过多会导致namenode元数据特别大, 占用太多内存，严重影响HDFS的性能
对 Hive 来说，在进行查询时，每个小文件都会当成一个块，启动一个Map任务来完成，而一个Map任务启动和初始化的时间远远大于逻辑处理的时间，就会造成很大的资源浪费。而且，同时可执行的Map数量是受限的
2.怎么解决小文件过多
1）使用 hive 自带的 concatenate 命令，自动合并小文件

#对于非分区表
alter table A concatenate;
#对于分区表
alter table B partition(day=20201224) concatenate;

2）调整参数减少Map数量

#执行Map前进行小文件合并
#CombineHiveInputFormat底层是 Hadoop的 CombineFileInputFormat 方法
#此方法是在mapper中将多个文件合成一个split作为输入
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; -- 默认

#每个Map最大输入大小(这个值决定了合并后文件的数量)
set mapred.max.split.size=256000000;   -- 256M

#一个节点上split的至少的大小(这个值决定了多个DataNode上的文件是否需要合并)
set mapred.min.split.size.per.node=100000000;  -- 100M

#一个交换机下split的至少的大小(这个值决定了多个交换机上的文件是否需要合并)
set mapred.min.split.size.per.rack=100000000;  -- 100M

设置map输出和reduce输出进行合并的相关参数：

#设置map端输出进行合并，默认为true
set hive.merge.mapfiles = true;

#设置reduce端输出进行合并，默认为false
set hive.merge.mapredfiles = true;

#设置合并文件的大小
set hive.merge.size.per.task = 256*1000*1000;   -- 256M

#当输出文件的平均大小小于该值时，启动一个独立的MapReduce任务进行文件merge
set hive.merge.smallfiles.avgsize=16000000;   -- 16M 

启用压缩

# hive的查询结果输出是否进行压缩
set hive.exec.compress.output=true;

# MapReduce Job的结果输出是否使用压缩
set mapreduce.output.fileoutputformat.compress=true;

减少reduce数量

#reduce 的个数决定了输出的文件的个数，所以可以调整reduce的个数控制hive表的文件数量，
#hive中的分区函数 distribute by 正好是控制MR中partition分区的，
#然后通过设置reduce的数量，结合分区函数让数据均衡的进入每个reduce即可。

#设置reduce的数量有两种方式，第一种是直接设置reduce个数
set mapreduce.job.reduces=10;

#第二种是设置每个reduce的大小，Hive会根据数据总大小猜测确定一个reduce个数
set hive.exec.reducers.bytes.per.reducer=5120000000; -- 默认是1G，设置为5G

#执行以下语句，将数据均衡的分配到reduce中
set mapreduce.job.reduces=10;
insert overwrite table A partition(dt)
select * from B
distribute by rand();

解释：如设置reduce数量为10，则使用 rand()， 随机生成一个数 x % 10 ，
这样数据就会随机进入 reduce 中，防止出现有的文件过大或过小

使用hadoop的archive将小文件归档

#用来控制归档是否可用
set hive.archive.enabled=true;
#通知Hive在创建归档时是否可以设置父目录
set hive.archive.har.parentdir.settable=true;
#控制需要归档文件的大小
set har.partfile.size=1099511627776;

#使用以下命令进行归档
ALTER TABLE A ARCHIVE PARTITION(dt='2022-02-24', hr='12');

#对已归档的分区恢复为原文件
ALTER TABLE A UNARCHIVE PARTITION(dt='2022-02-24', hr='12');

三、strict模式

开启严格模式对分区表进行查询，在where子句中没有加分区过滤的话，将禁止提交任务(默认：nonstrict)

set hive.mapred.mode=strict 开启严格模式

注：使用严格模式可以禁止以下三种类型的查询：

1. 对分区表的查询必须使用到分区相关的字段
分区表的数据量通常都比较大，对分区表的查询必须使用到分区相关的字段，不允许扫描所有分区，想想也是如果扫描所有分区的话那么对表进行分区还有什么意义呢。

当然某些特殊情况可能还是需要扫描所有分区，这个时候就需要记得确保严格模式被关闭。

2. order by必须带limit
因为要保证全局有序需要将所有的数据拉到一个Reducer上，当数据集比较大时速度会很慢。个人猜测可能是设置了limit N之后就会有一个很简单的优化算法：每个Reducer排序取N然后再合并排序取N即可，可大大减少数据传输量。

3. 禁止笛卡尔积查询(join必须有on连接条件)
Hive不会对where中的连接条件优化为on，所以join必须带有on连接条件，不允许两个表直接相乘。

四、并行执行优化

Hive会将一个查询转化成一个或者多个阶段。这样的阶段可以是MapReduce阶段、抽样阶段、合并阶段、limit阶段。或者Hive执行过程中可能需要的其他阶段。默认情况下，Hive一次只会执行一个阶段。不过，某个特定的job可能包含众多的阶段，而这些阶段可能并非完全互相依赖的，也就是说有些阶段是可以并行执行的，这样可能使得整个job的执行时间缩短。如果有更多的阶段可以并行执行，那么job可能就越快完成。

通过设置参数hive.exec.parallel值为true，就可以开启并发执行。在共享集群中，需要注意下，如果job中并行阶段增多，那么集群利用率就会增加。

set hive.exec.parallel=true; //打开任务并行执行
set hive.exec.parallel.thread.number=16; //同一个sql允许最大并行度，默认为8。

五、推测执行优化

在分布式集群环境下，因为程序bug(包括Hadoop本身的bug)，负载不均衡或者资源分布不均等原因，会造成同一个作业的多个任务之间运行速度不一致，有些任务的运行速度可能明显慢于其他任务(比如一个作业的某个任务进度只有50%，而其他所有任务已经运行完毕)，则这些任务会拖慢作业的整体执行进度。为了避免这种情况发生，Hadoop采用了推测执行(Speculative Execution)机制，它根据一定的法则推测出“拖后腿”的任务，并为这样的任务启动一个备份任务，让该任务与原始任务同时处理同一份数据，并最终选用最先成功运行完成任务的计算结果作为最终结果。

设置开启推测执行参数：Hadoop的mapred-site.xml文件中进行配置：

  mapreduce.map.speculative
  true
  If true, then multiple instances of some map tasks 
               may be executed in parallel.



  mapreduce.reduce.speculative
  true
  If true, then multiple instances of some reduce tasks 
               may be executed in parallel.

任务级别：set hive.mapred.reduce.tasks.speculative.execution=true

六、数据倾斜优化

set hive.map.aggr=true;  
set hive.groupby.skewindata = ture;

当选项设定为true时，生成的查询计划有两个MapReduce任务。

在第一个MapReduce中，map的输出结果集合会随机分布到reduce中，每个reduce做部分聚合操作，并输出结果。

这样处理的结果是，相同的Group By Key有可能分发到不同的reduce中，从而达到负载均衡的目的；

第二个MapReduce任务再根据预处理的数据结果按照Group By Key分布到reduce中(这个过程可以保证相同的Group By Key分布到同一个reduce中)，最后完成最终的聚合操作。

但是这个处理方案对于我们来说是个黑盒，无法把控。

那么在日常需求的情况下如何处理这种数据倾斜的情况呢；

sample采样，获取哪些集中的key；
将集中的key按照一定规则添加随机数；
进行join，由于打散了，所以数据倾斜避免了；
在处理结果中对之前的添加的随机数进行切分，变成原始的数据。