Hive中的数据倾斜优化

1. 基础优化

前提：
并行优化的前提是要服务器有足够的资源，如果没有资源则不会执行； 在SMBmapjoin中，资源不够导致运行很慢；

1.1 并行编译优化

Hive在同一时间只能编译一段HiveSQL，而工作环境有多个会话窗口，如果有多个HiveSQL并行执行，就会锁住，排队执行；需要开启并行执行；

开启：

1.2 并行执行优化

在运行SQL时，这个SQL的执行计划中，可能被拆分为多个 Stage 阶段，当各个阶段之间没有依赖关系的函数，可以尝试让多个阶段并行执行，提升执行效率；

带set的，即在会话窗口中配置，而没有set的在 Cloudera Manager中通用配置来配置；

1.3 小文件合并

小文件过多有什么影响？

MR角度：
每个文件单独切片，一个切片对应一个MapTask，会产生多个MapTask，每个MapTask都需要资源，而每个MapTask只处理很少的数据，导致资源浪费；

HDFS角度：
每个小文件都会有元数据，元数据太多，使得NameNode内存过大，一旦NameNode内存满了，即使DataNode有空间，也无法存数据！

开启：
是否开启map端小文件合并，适用于使用map没有reduce的时候，即map输出就是结果；
是否开启Reduce端文件合并操作；
合并后输出文件的最大值，默认128mb；
判定平均输出文件大小，当小于设置值时，认为出现了小文件问题，会按照最大值来进行合并；

1.4 矢量化查询

Hive默认查询执行引擎默认一次处理一行，而矢量化查询执行是一种hive特性，目的是按照每批 1024行读取数据！ 而且一次性对整个记录整合（而不是对单条记录）应用操作；

开启： set 即在会话中开启；

1.5 读取零拷贝 ZeroCopy

前提：ORC存储
在 ORC 存储格式下，在hive读取数据时，只需要读取跟SQL相关的列的数据即可，没有在SQL中使用的列，不进行读取！从而减少读取数据，提升效率；

开启：

总结：hive中很多优化都是围绕减少读取数据来进行的，比如读取零拷贝， 分区表，索引的优化…都是为了减少数据的扫描量，减少数据扫描量是显著的提升效率的方案；

2. 索引优化

2.1 原始索引

在【Hive3.0】以上已经不再使用 ×
原始索引不能自动更新！需要手动更新---- 重建索引，效率很差！

2.2 行组索引 row group（大小对比索引）-------用于数值查询

对所有的数据建索引；
ORC文件按行分段stripes ，每个片段又按照列存储；
每个stripe片段包含了每个列的 min、max索引数据，当查询时有< > =时，会根据min/max值，跳过扫描那些不包含的片段！

要求：
①必须是ORC的存储格式；
②在创建表的时候必须要开启行组索引 ；

③为了让行组有效，在向表中加载数据时，必须对需要使用索引的字段进行排序！否则min max就没有意义；
④主要应用在＞ ＜ = ，数值型数据；

例:

例：

2.3 布尔索引------用于等值查询 （=或in）

针对某一字段构建索引；
在对指定的字段建立索引时，会在【每个stripe片段中】，索引字段是相同的；
当查询条件中包含对字段的 =号过滤的时候，先从布尔索引获取一下是否包含该值，如果不包含则跳过stripe；

要求：
①必须是ORC格式存储
②对哪个字段进行等值查询，就将字段设置为布尔索引
③仅适合于等值查询，不局限于数值类型；
④建表时要指定哪些字段开启索引：

例：

例：

使用：
1）对于行组索引，建议常开
载入数据时，任意载入，在原始数据某些字段本来就是有序时，就可以使用行组索引；

2）主要对用于等值连接的字段开启索引即可，主要是指的是join的关联字段；

3. 数据倾斜优化

3.1 概述

何为数据倾斜

运行的过程当中，有多个Reduce，每个Reduce拿到的数据不均匀，其中有一个或几个拿到的数据远远大于其他reduce拿到的数量，此时任务出现了数据倾斜；

导致的问题

1. 执行效率下降（整个reduce时间，就看最后一个reduce结束时间）；
2. 由于reduce长时间运行不完，导致资源长期被占用，一旦超时，YARN会强制回收资源，导致运行失败
3. 导致节点宕机；

3.2 优化

group by 数据倾斜

• 方案一：Combiner预聚合
  一个MR，在【每一个MapTask】使用Combiner预聚合，将聚合之后的结果发往Reduce，这样在Reduce接收到的数据就少了，从而解决数据倾斜；

  配置：
  
  开启map端提前聚合 ；

• 方案二： 负载均衡（大Combiner） skew
  采用两个MR来解决，
  第一次MR负责将数据均匀落在不同的reduce上进行聚合统计，形成一个局部的结果；
  第二个MR，读取第一个MR的局部结果，按照相同的key发往同一个reduce，完成统计；

  配置：
  
  注意：
  当使用负载均衡时间，不支持在多列上去重，会报错；

join 数据倾斜

• 方案一： Map join、Bucket map join、 SMB map join
  将ruduce端的join操作移到到【MapTask的内存】，直接将倾斜排除，因为map一般不会有数据倾斜问题，map是根据切片来读取文件，切片block(128mb) 是均匀切的；

  但是三种map join都需要满足相关条件，如分桶的数量要整数倍/相同，分桶字段相同；但是很多时候不满足这些条件！

  注意：在map join之前，还有两个通用方法：
  ①空Key过滤：筛选掉无用字段
  ②空Key转换：null值太多时将其随机赋值打散，以免数据倾斜；nvl(n.id,rand())

• 方案二： 在Reduce端解决（无条件） skew join
  建议开启 union优化，以减少二次读写，减少union输出的额外扫描；整体效率更高；

  思路：
  将容易产生倾斜的key的值从整个环境中排除掉，将倾斜的数据单独找一个MR来处理，，算完后返回结果；
  如何知道哪个值导致的倾斜？
  ①编译（建表）期解决 （明确知道key值的倾斜问题时）
  明确知道key值有倾斜问题，一般采用编译器解决，在建表的时候，提前设置好对应值有倾斜即可，执行时，hive会将这些倾斜的key从这个MR排除，单独找一个MR来处理； 最后底层用union all合并；
  

  ②运行期解决**（当不知道哪个key导致倾斜时，设定阀值，动态检测）
  执行时，hive会记录每个key出现的次数，当key的次数达到特定的阀值，就认为key导致数据倾斜，将key的数据排除掉，单独用MR去处理；最后底层用union all合并；
  
  一般设置阀值为平均key个数的3~10倍，认为会产生数据倾斜；