Hive3.1.2——企业级调优

前言

本篇文章主要整理hive-3.1.2版本的企业调优经验，有误请指出~

一、性能评估和优化

1.1 Explain查询计划

使用explain命令可以分析查询计划，查看计划中的资源消耗情况，定位潜在的性能问题，并进行相应的优化。

explain执行计划见文章：

Hive调优——explain执行计划-CSDN博客文章浏览阅读843次，点赞18次，收藏11次。Hive调优——explain执行计划https://blog.csdn.net/SHWAITME/article/details/136092007?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522170790242216777224416146%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fblog.%2522%257D&request_id=170790242216777224416146&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~blog~first_rank_ecpm_v1~rank_v31_ecpm-1-136092007-null-null.nonecase&utm_term=explain&spm=1018.2226.3001.4450

1.2 调整并行度和资源配置

根据集群的配置和资源情况，合理调整Hive查询的并行度和资源分配，可以提高查询的并发性和整体性能。Hive在实现HQL计算运行时，会解析为多个Stage，有时候Stage彼此之间有依赖关系，只能挨个执行，但是在一些别的场景下，很多的Stage之间是没有依赖关系的。

例如Union语句，Join语句等等，这些Stage没有依赖关系，但是Hive依旧默认挨个执行每个Stage，这样会导致性能非常差，我们可以通过修改参数，开启并行执行，当多个Stage之间没有依赖关系时，允许多个Stage并行执行，提高性能。

-- 开启Stage并行化，默认为false
SET hive.exec.parallel=true;
-- 指定并行化线程数，默认为8
SET hive.exec.parallel.thread.number=16;

ps：调整并行度的措施，建议在数据量大，sql 逻辑复杂的时候使用。当数据量小或sql逻辑简单时开启并行度，优化效果不明显。

1.3 本地模式

使用hive的过程中，有一些数据量不大的表也会抓换成MapReduce处理，提交到yarn集群时，需要申请资源，等待资源分配，启动JVM进行，再运行Task，一系列的过程比较繁琐，严重影响性能。Hive为解决这个问题，延用了MapReduce中的设计，提供本地计算模式，允许程序不提交给yarn，直接在本地运行。

-- 开启本地模式
set hive.exec.mode.local.auto = true;

1.4 Fetch 抓取

Fetch 抓取是指：Hive 中对某些简单的查询可以不必使用 MapReduce 计算。 hive-default.xml.template 配置文件的 hive.fetch.task.conversion 默认是 more。此时在全局查找、字段查找、limit 查找等都不走mapreduce。例如：select * from employees;

二、Hive建表优化

2.1 分区表、分桶表

Hive的相关概念——分区表、分桶表-CSDN博客文章浏览阅读419次，点赞15次，收藏7次。Hive的相关概念——分区表、分桶表https://blog.csdn.net/SHWAITME/article/details/136111924?spm=1001.2014.3001.5502 总结：

分区表、分桶表不是建表的必要语法规则，是一种优化手段表，可选；
分区好处：用where进行分区过滤，查询指定分区的数据，避免全表扫描
分桶好处：基于分桶字段查询时，减少全表扫描；join时可以转换为SMB(Sort Merge Bucket join)
分区针对的是数据的存储路径；分桶针对的是数据文件（数据粒度更细）
分区字段不能是表中已经存在的字段，分桶的字段必须是表中已经存在的字段
分区字段是虚拟字段，其数据并不存储在底层的文件中；
分区字段值可以手动指定（静态分区），也可以根据查询结果位置自动推断（动态分区）
Hive支持多重分区，也就是说在分区的基础上继续分区，支持更细粒度的目录划分

2.2 文件格式及数据压缩优化

待补充~

三、HQL—Join优化

Hive Join的底层是通过MapReduce来实现的，Hive实现Join时，为了提高MapReduce任务的性能，提供了多种Join方案来实现。例如：适合小表Join大表的Map Join，大表Join大表的Reduce Join，以及大表Join的优化方案Bucket Join等。

3.1 Map Join

1）应用场景：小表join大表、小表Join小表

2）概述：Map Join是直接在Map阶段完成join工作，没有Shuffle阶段，从而避免了数据倾斜。

select /*+ mapjoin(b,c)*/  --mapjoin hint 定义小表，多个小表用逗号分隔
...
from t0 a
left join t1 b 
on a.id = b.id
left join t2 c
on a.id = c.id;

# MapJoin中多个小表用半角逗号（,）分隔，例如/*+ mapjoin(a,b,c)*/。

3）工作机制：使用hadoop中DistributedCache（分布式缓存）将小表广播到每个map任务节点，转换成哈希表加载到内存中，之后在mapper端和大表的分散数据做笛卡尔积，直接输出结果。

4）Map Join的特点：

要使用hadoop中的DistributedCache（分布式缓存）把小数据分布到各个计算节点，每个map节点都要把小数据库加载到内存，按关键字建立索引。
Map Join没有reduce任务，所以map直接输出结果，即有多少个map任务就会产生多少个结果文件
Hive3.1.2版本已经对Map Join进行了优化，小表放在左边和右边已经没有区别
MapJoin在Map阶段会将指定表的数据全部加载在内存中，因此指定的表仅能为小表且表被加载到内存后占用的总内存不得超过512 MB（默认）。由于MaxCompute是压缩存储，因此小表在被加载到内存后，数据大小会急剧膨胀。

5）参数设置：

#设置自动选择 Mapjoin，默认为true
set hive.auto.convert.join = true; 
#大表小表的阈值设置（默认25M以下认为是小表）
set hive.mapjoin.smalltable.filesize = 25*1000*1000;

3.2 Bucket Join

1）应用场景：大表Join大表

2）概述：将两张表按照相同的规则将数据划分、根据对应的规则的数据进行join、减少了比较次数，提高了性能

3.2.1 Bucket Join

语法：clustered by column
参数设置：set hive.optimize.bucketmapjoin = true;
要求：分桶字段 = Join字段，桶的个数相等或者成倍数

3.2.2 SMB Join

Sort Merge Bucket Join：基于有序的数据Join

语法：clustered by column sorted by (column )
参数设置：

set hive.optimize.bucketmapjoin = true;
set hive.auto.convert.sortmerge.join=true;
set hive.optimize.bucketmapjoin.sortedmerge = true;
set hive.auto.convert.sortmerge.join.noconditionaltask=true;

要求：分桶字段 = 排序字段= Join字段 ，两表的桶的个数相等或者成倍数
举例：

# 创建分桶表 bigtable_buck1
create table bigtable_buck1(
     id bigint,
     t bigint,
     uid string,
     keyword string,
     url_rank int,
     click_num int,
     click_url string
)
clustered by(id)
sorted by(id)
into 6 buckets
row format delimited fields terminated by '\t';
 
# 加载数据
load data local inpath '/opt/module/data/bigtable' into table 
bigtable_buck1;
 
 
#创建分桶表bigtable_buck2，分桶数和bigtable_buck1的分桶数为倍数关系
create table bigtable_buck2(
     id bigint,
     t bigint,
     uid string,
     keyword string,
     url_rank int,
     click_num int,
     click_url string
)
clustered by(id)
sorted by(id)
into 6 buckets
row format delimited fields terminated by '\t';
 
#加载数据
load data local inpath '/opt/module/data/bigtable' into table 
bigtable_buck2;

#================ SMB Join调优步骤

#设置参数
set hive.optimize.bucketmapjoin = true;
set hive.optimize.bucketmapjoin.sortedmerge = true;
set hive.input.format=org.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat;
 
 
# SMB Join
insert overwrite table jointable
select b.id,
       b.t,
       b.uid, 
       b.keyword, 
       b.url_rank, 
       b.click_num, 
       b.click_url
from bigtable_buck1 s
join bigtable_buck2 b
on b.id = s.id;

3.3 Reduce Join

1）应用场景：大表Join大表

2）概述：两张表的数据关联会经过shuffle阶段，Hive会自动判断是否满足Map Join，如果不满足Map Join，则自动执行Reduce Join

3）阶段阐述：

map端的主要工作：

生成键值对，以join on 条件中的列作为key，以join之后所关心的列作为value值，在value中还会包含Tag标记信息，用于标明此value对应哪张表

shuffle的主要工作：

根据key值进行hash分区, 按照hash值将键值对（key-value）发送到不同的reducer中

reduce端的主要工作：

Reducer通过Tag来识别不同的表中的数据，然后分别做合并操作

4）sql举例：

SELECT pageid, 
       age 
FROM page_view 
JOIN userinfo 
ON page_view.userid = userinfo.userid;

sql转化为mr任务流程如下图：

5）Reduce Join方法缺点：

map阶段没有对数据瘦身，shuffle的网络传输和排序性能很低。
reduce端需要通过Tag识别来源不同表的数据，很耗内存，容易导致OOM。

四、HQL—Group By

默认情况下， Map 阶段同一 Key 数据分发给一个 Reduce ，当一个 key数据过大时就可能发生数据倾斜。有些 聚合操作都可以先在 Map 端进行部分聚合(预先聚合)，最后在Reduce 端得出最终结果（归并merge）。开启 Map 端部预先聚合的参数如下：

（ 1）开启Map端聚合，默认为true

set hive.map.aggr = true;

（ 2 ）在 Map 端预先聚合操作的条数

set hive.groupby.mapaggr.checkinterval = 100000;

（ 3）数据倾斜时自动负载均衡（默认是 false ）

set hive.groupby.skewindata = true;

当该参数设定为 true时，生成的explain查询计划会有两个 MR：

第一个 MR Job 中，Map的输出结果会随机分发到Reduce中，每个Reduce做部分聚合操作并输出结果，这样处理的结果是相同的Group By Key有可能被分发到不同的Reduc中，从而达到负载均衡的目的；
第二个 MR Job将上一步预处理的数据结果按照 Group By Key 分布到 Reduce 中（这个过程可以保证相同的 Group By Key分布到同一个 Reduce 中），完成最终的聚合操作

五、HQL—非count(distinct) 去重

由于count ..distinct操作只启动一个reduce task执行，数据量大时要用 count ...group by替换count ..distinc去实现去重，但是需要注意group by可能会带来数据倾斜问题，解决方案见【四、HQL—Group By】

用group by 替换 count(distinct)的案例见文章：

Hive调优——count distinct替换-CSDN博客文章浏览阅读76次。Hive调优——count distinct替换https://blog.csdn.net/SHWAITME/article/details/136118294?spm=1001.2014.3001.5501

五、HQL—优化器引擎

5.1 CBO优化器

RBO（rule basic optimise）基于规则的优化器，根据设定好的规则来对程序进行优化；
CBO（cost basic optimise）基于代价的优化器，根据不同场景所需要付出的代价来合适选择优化的方案对数据的分布的信息【数值出现的次数，条数，分布】来综合判断用哪种处理的方案是最佳方案；Hive中支持RBO与CBO这两种引擎，默认使用的是RBO优化器引擎。

根据不同的应用场景，可以选择CBO，设置方式如下：

set hive.cbo.enable=true;
set hive.compute.query.using.stats=true;
set hive.stats.fetch.column.stats=true;
set hive.stats.fetch.partition.stats=true;

5.2 Analyze分析器

用于提前运行一个MapReduce程序，基于表或者分区的信息构建元数据信息【包含表的信息、分区信息、列的信息】，搭配CBO引擎一起使用。

-- 构建分区信息元数据
ANALYZE TABLE tablename
[PARTITION(partcol1[=val1], partcol2[=val2], ...)]
COMPUTE STATISTICS [noscan];

-- 构建列的元数据
ANALYZE TABLE tablename
[PARTITION(partcol1[=val1], partcol2[=val2], ...)]
COMPUTE STATISTICS FOR COLUMNS ( columns name1, columns name2...) [noscan];

-- 查看元数据
DESC FORMATTED [tablename] [columnname];

--分析优化器
--构建表中分区数据的元数据信息
ANALYZE TABLE tb_login_part PARTITION(logindate) COMPUTE STATISTICS;
--构建表中列的数据的元数据信息
ANALYZE TABLE tb_login_part  COMPUTE STATISTICS FOR COLUMNS  userid;
--查看构建的列的元数据
desc formatted tb_login_part userid;

六、HQL—谓词下推（PPD）

谓词下推Predicate Pushdown（PPD）：在不影响结果的情况下，尽量将过滤条件提前执行。谓词下推后，过滤条件在map端执行，减少了map端的输出，降低了数据在集群上传输的量，提升任务性能。

谓词下推的场景分析见文章：

Hive的Join连接、谓词下推-CSDN博客文章浏览阅读502次，点赞11次，收藏10次。Hive的Join连接https://blog.csdn.net/SHWAITME/article/details/136105973?spm=1001.2014.3001.5501

七、HQL—in/exists 语句

in/exists操作，推荐使用Hive的left semi join（左半连接）进行替代

# in / exists 实现
select a.id, a.name from a where a.id in (select b.id from b);
select a.id, a.name from a where exists (select id from b where a.id =
b.id);

#可以使用 join 来改写
select a.id, a.name from a join b on a.id = b.id;

#left semi join 实现
select a.id, a.name from a left semi join b on a.id = b.id;

left semi join（左半连接）的详细说明见文章：

hive/spark--left semi/anti join_sparksql left semi join-CSDN博客

八、HQL—CTE 公共表达式

拖慢HQL查询效率的原因除了join引发的shuffle过程外，还有一个就是子查询调用次数较多，存在冗余代码块。因此我们可以借助CTE 公共表达式，简单来讲就是：with as 语句，将代码中的子查询事先提取出来（类似临时表），可以避免重复计算。

#==============优化前
select
    uid,
    --每个用户一月份的订单数
    sum(if(dt = '2018-01', 1, 0)) as                 m1_count,
    --每个用户二月份的订单数
    sum(if(dt = '2018-02', 1, 0)) as                 m2_count,
    --每个用户三月份的订单数(当月订单金额超过10元的订单个数)
    sum(if(dt = '2018-03' and oamount > 10, 1, 0))   m3_count,
    --当月(3月份)首次下单的金额
    sum(if(dt = '2018-03' and rk = 1, oamount, 0))   m3_first_amount,
    --当月(3月份)末次下单的金额(rk =cnt小技巧)
    sum(if(dt = '2018-03' and rk = cnt, oamount, 0)) m3_last_amount
from (
         select
             oid,
             uid,
             otime,
             date_format(otime, 'yyyy-MM') as                                                  dt,
             oamount,
             ---计算rk的目的是为了获取记录中的第一条
             row_number() over (partition by uid,date_format(otime, 'yyyy-MM') order by otime) rk,
             --- 计算cnt的目的是为了获取记录中的最后一条
             count(*) over (partition by uid,date_format(otime, 'yyyy-MM'))                    cnt
         from t_order
         order by uid
     ) tmp
group by uid
having m1_count > 0
   and m2_count = 0;


#================优化后
with tmp as (
    select
        oid,
        uid,
        otime,
        date_format(otime, 'yyyy-MM') as                                                  dt,
        oamount,
        ---计算rk的目的是为了获取记录中的第一条
        row_number() over (partition by uid,date_format(otime, 'yyyy-MM') order by otime) rk,
        --- 计算cnt的目的是为了获取记录中的最后一条
        count(*) over (partition by uid,date_format(otime, 'yyyy-MM'))                    cnt
    from t_order
    order by uid
)
select
    uid,
    --每个用户一月份的订单数
    sum(if(dt = '2018-01', 1, 0)) as                 m1_count,
    --每个用户二月份的订单数
    sum(if(dt = '2018-02', 1, 0)) as                 m2_count,
    --每个用户三月份的订单数(当月订单金额超过10元的订单个数)
    sum(if(dt = '2018-03' and oamount > 10, 1, 0))   m3_count,
    --当月(3月份)首次下单的金额
    sum(if(dt = '2018-03' and rk = 1, oamount, 0))   m3_first_amount,
    --当月(3月份)末次下单的金额(rk =cnt小技巧)
    sum(if(dt = '2018-03' and rk = cnt, oamount, 0))  m3_last_amount
from tmp
group by uid
having m1_count >0 and m2_count=0;

ps：hive中的CTE公共表达式文章：Hive的CTE 公共表达式-CSDN博客文章浏览阅读273次，点赞5次，收藏3次。Hive的CTE 公共表达式https://blog.csdn.net/SHWAITME/article/details/136108359?spm=1001.2014.3001.5501

九、合理设置 Map 及 Reduce 数

1）通常情况下，作业会通过input的目录产生一个或者多个map 任务，map数量主要的决定因素有：input 的文件总个数，input 的文件大小，集群设置的文件块大小。

2 ）map数并非越多越好

如果一个任务有很多小文件（远远小于块大小 128m ），则每个小文件会启动一个 map 任务来执行，而一个 map 任务启动和初始化的时间远远大于逻辑处理的时间，造成很大的资源浪费。

9.1 复杂文件增加Map数

当input的文件很大，任务逻辑复杂，map执行非常慢的时候，可以考虑增加Map数，公式：

computeSliteSize(Math.max(minSize,Math.min(maxSize,blocksize)))

让maxSize低于blocksize，此时公式 computeSliteSize(Math.max(minSize,Math.min(maxSize,blocksize))) = maxSize

原map个数= 输入文件的数据量 / computeSliteSize = 输入文件的数据量 / blocksize， 现在map个数=输入文件的数据量 / computeSliteSize = 输入文件的数据量 / maxSize

9.2 合理设置 Reduce 数

9.2.1 调整reduce数的方法一

# 每个reduce任务处理的数据量默认是 256MB
hive.exec.reducers.bytes.per.reducer=256*1000*1000
# 整个MR任务支持开启的reduce数的上限值，默认为1009个
hive.exec.reducers.max=1009

reducer数的计算公式：

个数N = min ( hive.exec.reducers.max ，总输入数据量 / hive.exec.reducers.bytes.per.reducer)

根据该公式可以得知：reduce个数（hdfs上的落地文件数量）是动态计算的

9.2.2 调整reduce数的方法二

在hadoop的mapred-default.xml 文件中修改下列参数，

#设置MR job的Reduce总个数
set mapreduce.job.reduces = 15;

ps：reduce个数并不是越多越好

过多的启动和初始化reduce也会消耗时间和资源；
有多少个reduce就会有多少个输出文件，如果生成了很多个小文件，那么这些小文件作为下一个任务的输入时，则也会出现小文件过多的问题；
在设置 reduce 个数时还需要考虑：单个reduce 任务处理数据量大小要合适（避免数据倾斜）

十、Hive的小文件合并

Hive小文件问题及解决方案，见文章：
Hive的小文件问题-CSDN博客文章浏览阅读452次，点赞7次，收藏12次。Hive的小文件问题https://blog.csdn.net/SHWAITME/article/details/136108785?spm=1001.2014.3001.5501