Hive学习---7、企业级调优（一）

1、企业级调优

1.1 计算资源配置

到此学习的计算环境为HIve on MR。计算资源的调整主要包括Yarn和MR。

1.1.1 Yarn资源配置

1、Yarn配置说明

需要调整的Yarn的参数均与CPU、内存等资源有关，核心配置参数如下：

（1）yarn.nodemanager.resource.memory-mb

该参数的含义是，一个NodeManager节点分配给Container使用的内存。该参数的配置，取决于NodeManager所在节点的总内存容量和该节点运行的其他服务的数量。

（2）yarn.nodemanager.resource.cpu-vcores

该参数的含义是，一个NodeManager节点分配给Container使用的CPU核数。该参数的配置，同样取决于NodeManager所在节点的总CPU核数和该节点运行的其他服务。

（3）yarn.scheduler.maximum-allocation-mb

该参数的含义是，单个Container能够使用的最大内存。

（4）yarn.scheduler.minimum-allocation-mb

该参数的含义是，单个Container能够使用的最小内存。

1.1.2 MapReduce资源配置

MapReduce资源配置主要包括Map Task的内存和CPU核数，以及Reduce Task的内存和CPU核数。核心配置参数如下：

1、mapreduce.map.memory.mb

该参数的含义是，单个Map Task申请的container容器内存大小，其默认值为1024。该值不能超出yarn.scheduler.maximum-allocation-mb和yarn.scheduler.minimum-allocation-mb规定的范围。

2、mapreduce.map.cpu.vcores

该参数的含义是，单个Map Task申请的container容器cpu核数，其默认值为1。该值一般无需调整。

3、mapreduce.reduce.memory.mb

该参数的含义是，单个Reduce Task申请的container容器内存大小，其默认值为1024。该值同样不能超出yarn.scheduler.maximum-allocation-mb和yarn.scheduler.minimum-allocation-mb规定的范围。

4、mapreduce.reduce.cpu.vcores

该参数的含义是，单个Reduce Task申请的container容器cpu核数，其默认值为1。该值一般无需调整。

1.2 Explain查看执行计划

1.2.1 Explain执行计划概述

Explain呈现的执行计划，由一系列Stage组成，这一系列Stage具有依赖关系，每个Stage对应一个MapReduce Job，或者一个文件系统操作。

若某个Satge对应的一个MapReduce Job，其Map段和Reduce端的计算逻辑分别由Map Operator Tree 和Reduce Operator Tree进行描述，Operator Tree 由一系列的Operator组成，应该Operator代表在Map或Reduce阶段的一个单一的逻辑操作，例如TableScan Operator，Select Operator，Join Operator等。

下图是由一个执行计划绘制而成：

常见的Operator及作用如下：

TableScan：表扫描操作，通常map段第一个操作肯定是表扫描操作

Select Operator：选取操作

Group by Operator：分组聚合操作

Reduce Output Operator：输出到reduce操作

Filter Operator：过滤操作

Join Operator：join操作

File Output Operator：文件输出操作

Fetch Operator：客户端获取数据操作

1.2.2 基本语法

Explain [Formatted | Extended | Dependency] query-sql

注：Formatted、Extended、Dependency关键字为可选项，各自作用如下。

FORMATTED：将执行计划以JSON字符串的形式输出

EXTENDED：输出执行计划中的额外信息，通常是读写的文件名等信息

DEPENDENCY：输出执行计划读取的表及分区

1.2.3 案例实操

1、查看下面这条语句的执行

explain formatted 
select
    sku_id,
    count(*)
from order_detail
group by sku_id; 

2、执行计划如下图

1.3 HQL语法优化之分组聚合优化

1.3.1 优化说明

Hive中未经优化的分组聚合是通过一个MapReduce Job实现的。Map端负责读取数据，并按照分组字段分区，通过Shuffle，将数据发往Reduce端，各组数据在Reduce端完成最终的聚合运算。

Hive对分组聚合的优化主要是围绕着减少Shuffle数据量进行，具体做法是map-side聚合。所谓map-side聚合，就是在map端维护一个hash table，利用其完成部分的聚合，然后将部分聚合的结果，按照分组字段分区，发送至reduce端，完成最终的聚合。map-side聚合能有效减少Shuffle的数据量，提高分组聚合运算的效率。

map-side聚合相关的参数如下：

--启动map-side聚合
set hive.map.aggr=true;
--用于检测源表数据是否适合进行map-side聚合。检测的方法是：先对若干条数据进行map-side聚合，若聚合后的条数和聚合前的条数比值小于该值，则认为该表适合进行map-side聚合；否则，认为该表数据不适合进行map-side聚合，后续数据便不再进行map-side聚合。
set hive.map.aggr.hash.min.reduction=0.5;
--用于检查源表是否适合map-side聚合的条数
set hive.groupby.mapaggr.checkinterval=100000;

1.3.1 优化案例

1、实例SQL

select
    product_id,
    count(*)
from order_detail
group by product_id;

2、优化前

3、优化思路

可以考虑开启map-side聚合，配置以下参数

--启用map-side聚合，默认是true
set hive.map.aggr=true;
--用于检测源表数据是否适合进行map-side聚合。检测的方法是：先对若干条数据进行map-side聚合，若聚合后的条数和聚合前的条数比值小于该值，则认为该表适合进行map-side聚合；否则，认为该表数据不适合进行map-side聚合，后续数据便不再进行map-side聚合。
set hive.map.aggr.hash.min.reduction=0.5;
--用于检测源表是否适合map-side聚合的条数。
set hive.groupby.mapaggr.checkinterval=100000;
--map-side聚合所用的hash table，占用map task堆内存的最大比例，若超出该值，则会对hash table进行一次flush。
set hive.map.aggr.hash.force.flush.memory.threshold=0.9;

1.4 HQL语法优化之Join优化

1.4.1 Join算法概述

Hive拥有多种Join算法，包括Common Join，Map Join，Bucket Map Join,Sort Merge Bucket Map Join等，下面对每种Join算法做简要说明：

1、Common Join

Common Join算法是Hive中最稳定的算法，其通过一个MapReduce Job完成一个Join操作。Map端负责读取Join操作所需表的数据，并按照关联字段进行分区，通过Shuffle，将其发送到Reduce端，相同Key的数据在Reduce端完成最终的Join操作。如下图所示：

需要注意的是，sql语句中的join操作和执行计划中的Common Join任务并非一对一的关系，一个sql语句中的相邻的且关联字段相同的多个join操作可以合并为一个Common Join任务。

例如：

hive (default)> 
select 
    a.val, 
    b.val, 
    c.val 
from a 
join b on (a.key = b.key1) 
join c on (c.key = b.key1)

上述sql语句中两个join操作的关联字段均为b表的key1字段，则该语句中的两个join操作可由一个Common Join任务实现，也就是可通过一个Map Reduce任务实现。

select 
    a.val, 
    b.val, 
    c.val 
from a 
join b on (a.key = b.key1) 
join c on (c.key = b.key2)

上述sql语句中的两个join操作关联字段各不相同，则该语句的两个join操作需要各自通过一个Common Join任务实现，也就是通过两个Map Reduce任务实现。

2、Map Join

Map Join算法可以通过两个只有map阶段的Job完成一个join操作。其适用场景为大表join小表。若某join操作满足操作要求，则第一个job会读取小表的数据，将其制作为hashtable，并上传到hadoop分布式缓存（本质上是上传至HDFS）。第二个Job会先从分布式缓存中读取小表数据，并缓存在Map Task的内存中，然后扫描大表数据，这样在map端即可完成关联操作。如下图所示：

3、Bucket Map Join

Bucket Map Join 是对Map Join算法的改进，其打破了Map Join只适用于大表join小表的限制，可用于大表join大表的场景。

Bucket Map Join 的核心思想是：若能保证参与Join的表均为分桶表，且关联字段为分桶字段，且其中一张表的分桶数量是另外一张分桶表数量的整数倍，就能保证参与Join的两张表的分桶之间具有明确的关联关系，所以就可以在两表的分桶间进行Map Join操作了。这样一来，第二个Job的Map端就无需再缓存小表的全表数据了，而只需缓存其所需要的分桶即可。其原理如图所示：

4、Sort Merge Bucket Map Join

Sort Merge Bucket Map Join（简称SMB Map Join）基于Bucket Map Join。SMB Map Join要求，参与join的表均为分桶表，且需保证分桶内的数据是有序的，且分桶字段、排序字段和关联字段为相同字段，且其中一张表的分桶数量是另外一张表分桶数量的整数倍。

SMB Map Join同Bucket Join一样，同样是利用两表各分桶之间的关联关系，在分桶之间进行join操作，不同的是，分桶之间的join操作的实现原理。Bucket Map Join，两个分桶之间的join实现原理为Hash Join算法；而SMB Map Join，两个分桶之间的join实现原理为Sort Merge Join算法。

Hash Join和Sort Merge Join均为关系型数据库中常见的Join实现算法。Hash Join的原理相对简单，就是对参与join的一张表构建hash table，然后扫描另外一张表，然后进行逐行匹配。Sort Merge Join需要在两张按照关联字段排好序的表中进行，其原理如图所示：

Hive中的SMB Map Join就是对两个分桶的数据按照上述思路进行Join操作。可以看出，SMB Map Join与Bucket Map Join相比，在进行Join操作时，Map端是无需对整个Bucket构建hash table，也无需在Map端缓存整个Bucket数据的，每个Mapper只需按顺序逐个key读取两个分桶的数据进行join即可。

1.4.2 Map Join

优化说明：

Map Join有两种触发方式，一种是用户在SQL语句中增加hint提示，另一种是Hive优化器根据参与Join表的数据量大小，自动触发。

1、hint提示

用户可以通过如下方式，指定通过map join算法，并且ta将作为map join中的小表。这种方式已经过时，不推荐使用。

select /*+ mapjoin(ta) */
    ta.id,
    tb.id
from table_a ta
join table_b tb
on ta.id=tb.id;

2、自动触发

Hive在编译SQL语句阶段，起初所有的join操作均采用Common Join算法实现。

之后在物理优化阶段，Hive会根据每个Common Join任务所需表的大小判断该Common Join任务是否能够转换为Map Join任务，若满足要求，便将Common Join任务自动转换为Map Join任务。

但有些Common Join任务所需的表大小，在SQL的编译阶段是未知的（例如对子查询进行join操作），所以这种Common Join任务是否能转换成Map Join任务在编译阶是无法确定的。

针对这种情况，Hive会在编译阶段生成一个条件任务（Conditional Task），其下会包含一个计划列表，计划列表中包含转换后的Map Join任务以及原有的Common Join任务。最终具体采用哪个计划，是在运行时决定的。大致思路如下图所示：

Map join自动转换的具体判断逻辑如下图所示：

图中涉及的参数如下：

--启动Map Join自动转换
set hive.auto.convert.join=true;
--一个Common Join operator转为Map Join operator的判断条件,若该Common Join相关的表中,存在n-1张表的已知大小总和<=该值,则生成一个Map Join计划,此时可能存在多种n-1张表的组合均满足该条件,则hive会为每种满足条件的组合均生成一个Map Join计划,同时还会保留原有的Common Join计划作为后备(back up)计划,实际运行时,优先执行Map Join计划，若不能执行成功，则启动Common Join后备计划。
set hive.mapjoin.smalltable.filesize=250000;
--开启无条件转Map Join
set hive.auto.convert.join.noconditionaltask=true;
--无条件转Map Join时的小表之和阈值,若一个Common Join operator相关的表中，存在n-1张表的大小总和<=该值,此时hive便不会再为每种n-1张表的组合均生成Map Join计划,同时也不会保留Common Join作为后备计划。而是只生成一个最优的Map Join计划。
set hive.auto.convert.join.noconditionaltask.size=10000000;