Hive 优化：Hive 的执行计划、Bucketing、MapJoin、数据倾斜

1. hive的执行计划

1.1 为什么使用EXPLAIN

在处理大数据查询时，性能是一个关键问题。复杂的查询可能涉及多个表的连接、过滤、分组和排序操作，这些操作如果处理不当，会导致查询效率低下，甚至无法完成。在Hive中，EXPLAIN命令可以帮助我们理解查询的执行计划，识别出潜在的性能瓶颈，从而进行有针对性的优化。

1.2 使用EXPLAIN的步骤

使用EXPLAIN命令只需在查询语句前加上EXPLAIN关键字，即可查看该查询的执行计划。通过EXPLAIN输出的执行计划，可以看到每个阶段的操作，如表扫描、数据过滤、分组聚合等。

• 编写SQL查询：确定要执行的查询。
• 添加EXPLAIN关键字：在查询面前加上EXPLAIN

EXPLAIN SELECT ...;

1.3 EXPLAIN在什么场合使用

EXPLAIN命令适用于以下场合：

• 复杂查询：当查询涉及多个表的连接、复杂的过滤条件或分组聚合时，使用EXPLAIN可以帮助理解查询的执行步骤，并能找出性能瓶颈。
• 查询优化：在优化查询性能时，EXPLAIN可以帮助验证优化的效果。
• 调试查询：如果查询执行时间过长或结果不符合预期，使用EXPLAIN可以帮助诊断问题，发现潜在的错误或低效的操作。

2. 分桶

2.1 为什么要使用分桶

• 为了抽样
  在处理大规模数据集时，在开发和修改查询的阶段，可以使用整个数据集的一部分进行抽样测试查询、修改。可以使得开发更高效。
• 为了提高查询
  连接两个在（包含连接列的）相同列上划分了桶的表，可以使用Map端连接（Map-side join）高效的实现。
  比如join操作，对于join操作两个表有一个相同的列，如果对这两个表都进行了桶操作。那么将保存相同列值的桶进行join操作就可以，可以大大减少join的数据量。

3. Map Join

3.1 Map Join

Map Join就是在Map阶段进行表之间的连接。而不需要进入到Reduce阶段才进行连接。这样就节省了在shuffle阶段时要进行大量的数据传输，从而起到了优化作业的作用。
要使MapJoin能够顺利进行，就必须满足这样的条件：除了一份表的数据分布在不同的Map中外，其他连接的表的数据必须在每个Map中有完整拷贝。
所以并不是所有的场景都适合用Map join。它通常会用在如下的一些场景：在两个要连接的表中，有一个大表，有一个小表，这个小表可以存放在内存中而不影响性能。
这样就可以把小表文件复制到每一个Map任务的本地，再让Map把文件读到内存中待用。

3.1.1 大小表关联

select f.a,f.b from A t join B f on (f.a = t.a and f.ftime=20110802)

该语句中B表有30亿行记录，A表只有100行记录，而且B表中数据倾斜特别严重，有一个key上有15亿行记录，在运行过程中特别的慢，而且在reduece的过程中遇到执行时间过长或者内存不够的问题。
MAPJION会把小表全部读入内存中，在map阶段直接拿另外一个表的数据和内存中表数据做匹配，由于在map时进行了join操作，省去了reduce运行的效率会高很多。
这样就不会由于数据倾斜导致某个reduce上落数据太多而失败。于是原来的sql可以通过使用hint的方式指定join时使用mapjoin。
注意：
在实际使用中，只要根据业务调整小表的阈值即可，hive会自动帮我们完成mapjoin，提高执行的效率。

3.1.2 不等连接

map join还有一个很大的好处是能够进行不等连接的操作，如果将不等条件写在where中，那么mapreduce过程中会进行笛卡尔积，运行效率特别低，如果使用map join操作，在map的过程中就完成了不等值的join操作，效率会高很多。

select A.a ,A.b from A join B where A.a>B.a

3.2 Bucket-MapJoin

3.2.1 作用

两个表join的时候，小表不足以放到内存中，但是又想用map side join这个时候就要用到Bucket MapJoin。其方法是两个join表在join key上都做hash bucket，并且把你打算复制的那个（相对）小表的bucket数设置为大表的倍数。这样数据就会按照key join，做hash bucket。小表依然复制到所有节点，Map join的时候，小表的每一组bucket加载成hashtable，与对应的一个大表bucket做局部join，这样每次只需要加载部分hashtable就可以了。

3.2.2 条件

1） set hive.optimize.bucketmapjoin = true;
2） 一个表的bucket数是另一个表bucket数的整数倍
3） bucket列 == join列(数值类型)
4） 必须是应用在map join的场景中

注意：如果表不是bucket的，则只是做普通join。

3.3 SMB Join

全称Sort Merge Bucket Join。

3.3.1 作用

大表对小表应该使用Map Join来进行优化，但是如果是大表对大表，如果进行shuffle，就会非常慢，并且容易出现异常，此时就可以使用SMB Join来提高性能。SMB Join基于Bucket-MapJoin的有序bucket，可实现map端完成join操作，可以有效地减少或避免shuffle的数据量。SMB join的条件和Map join类似但又不同。

bucket mapjoin	SMB join
set hive.optimize.bucketmapjoin = true;	set hive.optimize.bucketmapjoin = true; set hive.optimize.bucketmapjoin.sortedmerge = true; set hive.auto.convert.sortmerge.join=true; set hive.auto.convert.sortmerge.join.noconditionaltask=true;
一个表的bucket数是另一个表bucket数的整数倍	小表的bucket数=大表bucket数
bucket列 == join列	Bucket 列 == Join 列(数值) == sort 列
必须是应用在map join的场景中	必须是应用在bucket mapjoin 的场景中

4. 数据倾斜

4.1 表连接数据倾斜(Join skew in)

4.1.1 运行时优化

set hive.optimize.skewjoin=true;
默认关闭。
如果大表和大表进行join操作，则可采用skewjoin(倾斜关联)来开启对倾斜数据的优化。
skewjoin原理：

1. 对于skewjoin.key，在执行job时，将它们存入临时的HDFS目录，其它数据正常执行
2. 对倾斜数据开启map join操作（多个map并行处理），对非倾斜值采取普通join操作
3. 将倾斜数据集和非倾斜数据集进行合并Union操作。

开启skewin以后，究竟多大的数据才会被认为是倾斜了的数据呢？
set hive.skewjoin.key=100000;
默认值100000。
如果join的key对应的记录条数超过这个值，就认为这个key产生了数据倾斜，则会对其进行分拆优化。

4.1.2 编译时优化

上面的配置项其实应该理解为hive.optimize.skewjoin.runtime，也就是sql运行时来对偏斜信息进行优化；除此之外还有另外一个配置：
set hive.optimize.skewjoin.compiletime=true;
默认关闭。
此参数的用处和上面的hive.optimize.skewjoin一致，但在编译sql时就已经将执行计划优化完毕。但要注意的是，只有在表的元数据中存储的有数据倾斜信息时，才能生效。因此建议runtime和compiletime都设置为true。
可以通过建表语句来指定数据倾斜元数据：

CREATE TABLE list_bucket_single (key STRING, value STRING)
-- 倾斜的字段和需要拆分的key值
SKEWED BY (key) ON (1,5,6)
--  为倾斜值创建子目录单独存放
[STORED AS DIRECTORIES];

4.2 分组统计数据倾斜(Groupby skew in)

4.2.1 Map阶段聚合

set hive.map.aggr=true;

开启map端combiner。此配置可以在group by语句中提高HiveQL聚合的执行性能。这个设置可以将顶层的聚合操作放在Map阶段执行，从而减轻数据传输和Reduce阶段的执行时间，提升总体性能。默认开启，无需显示声明。

4.2.2 MRJob随机数打散

set hive.groupby.skewindata=true;

默认关闭。
这个配置项是用于决定group by操作是否支持倾斜数据的负载均衡处理。当数据出现倾斜时，如果该变量设置为true，那么Hive会自动进行负载均衡。
当选项设定为 true，生成的查询计划会有两个 MR Job。
第一个MR Job中，Map 的输出结果集合会随机分布到Reduce中，每个Reduce做部分聚合操作，并输出结果，这样处理的结果是相同的Group By Key有可能被分发到不同的Reduce中，从而达到负载均衡的目的；
第二个MR Job再根据预处理的数据结果按照Group By Key分布到Reduce中（这个过程可以保证相同的Group By Key被分布到同一个Reduce中），最后完成最终的聚合操作。

注意：在多个列上进行的去重操作与hive环境变量hive.groupby.skewindata存在冲突。
当hive.groupby.skewindata=true时，hive不支持多列上的去重操作，并报错：

Error in semantic analysis: DISTINCT on different columns not supported with skew in data.