Hive第三讲 调优

hive 调优

1.调优概述1、Hive的建表设计层面

1.1利用分区表优化1.2 利用分桶表优化1.3 选择合适的文件存储格式1.4 选择合适的压缩格式 2、HQL语法和运行参数层面

2.1.查看Hive执行计划2.2 列裁剪2.3 分区裁剪2.4 谓词下推2.5 合并小文件2.6 join优化2.7 启用 MapJoin2.8 CBO优化2.9 group by 优化2.10 Count Distinct优化 3、Hive架构层面

3.1 启用本地抓取3.2 本地执行优化3.3 JVM重用3.4 并行执行3.5 推测执行3.6 严格模式 4.调优实战

4.1 第一个例子：日志表和用户表做链接4.2 第二个例子：位图法求连续七天发朋友圈的用户

1.调优概述

Hive调优的作用：在保证业务结果不变的前提下，降低资源的使用量，减少任务的执行时间。

1、Hive的建表设计层面 1.1利用分区表优化

当一个 Hive 表的查询大多数情况下，会根据某一个字段进行筛选时，那么非常适合创建为分区表，该字段即为分区字段。(避免全表扫描)

1.2 利用分桶表优化

分桶：默认规则：Hash散列,一个分桶中会有多个不同的值,如果一个分桶中，包含了某个值，这个值的所有记录，必然都在这个分桶。
分桶的语法：
CLUSTERED BY(userid) SORTED BY(viewTime) INTO 32 BUCKETS
CLUSTERED BY(userid) 表示按照 userid 来分桶
SORTED BY(viewTime) 按照viewtime来进行桶内排序
INTO 32 BUCKETS 分成多少个桶。
两个表以相同方式（相同字段）划分桶，两个表的桶个数是倍数关系。
1、采样

分桶抽样：
select * from student tablesample(bucket 3 out of 32);
随机采样：rand() 函数
select * from student order by rand() limit 100; // 效率低
select * from student distribute by rand() sort by rand() limit 100; // 推荐使
用这种
数据块抽样：tablesample()函数
select * from student tablesample(10 percent); # 百分比
select * from student tablesample(5 rows); # 行数
select * from student tablesample(5 M); # 大小

2、join 条件(这样保证两张表中相同的分桶编号的数据，在同一个节点)。

1.3 选择合适的文件存储格式

TextFile是最简单的存储格式，它是纯文本记录，也是Hive的默认格式。
创建表时，特别是宽表，尽量使用 ORC、ParquetFile 这些列式存储格式
ORC File：
存储方式：数据按行分块，每块按照列存储
Parquet File :
存储方式：列式存储。
Parquet对于大型查询的类型是高效的。默认Snappy,支持Impala 查询引擎

1.4 选择合适的压缩格式

1.压缩算法:使用于cpu的压力不大，io密集型的数据。
压缩速度:
压缩解压速度:

2、HQL语法和运行参数层面 2.1.查看Hive执行计划

## 查看执行计划，添加extended关键字可以查看更加详细的执行计划
explain [extended] query

2.2 列裁剪

列裁剪就是在查询时只读取需要的列

2.3 分区裁剪

set hive.optimize.pruner=true; ## 默认是true

分区裁剪就是只读取需要的分区

2.4 谓词下推

将 SQL 语句中的 where 谓词逻辑都尽可能提前执行，减少下游处理的数据量。
set hive.optimize.ppd=true; ## 默认是true

select a.*, b.* from a join b on a.id = b.id where b.age > 20;
select a.*, c.* from a join (select * from b where age > 20) c on a.id = c.id

2.5 合并小文件

如果一个mapreduce job碰到一堆小文件作为输入，一个小文件启动一个Task
Map 输入合并
默认的是textinputformat

## Map端输入、合并文件之后按照block的大小分割（默认）
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;
## Map端输入，不合并
set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;

2.6 join优化

1、优先过滤后再进行Join操作，最大限度的减少参与join的数据量
2、小表join大表，最好启动mapjoin，hive自动启用mapjoin, 小表不能超过25M，可以更改.

2.7 启用 MapJoin

大表 join 小表 小表满足需求： 小表数据小于控制条件时

## hive会基于表的size自动的将普通join转换成mapjoin
set hive.auto.convert.join.noconditionaltask=true;

在维度建模数据仓库中，事实表就是 probe table，维度表就是 build table。这种 Join 方式在 map 端直接完成 join 过程.

// 在没有开启mapjoin的情况下，执行的是reduceJoin
SELECt  smallTable.key, bigTable.value FROM
smallTable JOIN bigTable ON smallTable.key = bigTable.key;

Sort-Merge-Bucket(SMB) Map Join
它是另一种Hive Join优化技术，使用这个技术的前提是所有的表都必须是分桶表（bucket）和分桶排序的（sort）。分桶表的优化.

2.8 CBO优化

select a.*, b.*, c.* from a join b on a.id = b.id join c on a.id = c.id;
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;

2.9 group by 优化

Map 阶段同一个 Key 的数据会分发到一个 Reduce 上，当一个 Key 的数据过大时会产生数据倾斜。进行 group by 操作时可以从以下两个方面进行优化：

Map端部分聚合

## 开启Map端聚合参数设置
set hive.map.aggr=true;
# 设置map端预聚合的行数阈值，超过该值就会分拆job，默认值100000
set hive.groupby.mapaggr.checkinterval=100000

2.有数据倾斜时进行负载均衡

# 自动优化，有数据倾斜的时候进行负载均衡（默认是false）
set hive.groupby.skewindata=false;

当选项设定为 true 时，生成的查询计划有两个 MapReduce 任务。
两阶段聚合:
在第一个 MapReduce 任务中，map 的输出结果会随机分布到 reduce 中做部分聚合。
第二个 MapReduce 任务再根据预处理的数据结果按照 group by key 分布到各个 reduce 中，最后完成最终的聚合操作。

2.10 Count Distinct优化

count(distinct) 优化
select count(distinct id) from tablename;

-- 先 group by 在 count
select count(1) from (
select age from student
where department >= "MA"
group by age
) t;

3、Hive架构层面 3.1 启用本地抓取

Hive 从 HDFS 中读取数据，有两种方式：启用MapReduce读取 和 直接抓取
hive.fetch.task.conversion

minimal：只有 select * 、在分区字段上 where 过滤、有 limit 这三种场景下才启用直接抓取方
式。
more：在 select、where 筛选、limit 时，都启用直接抓取方式。

3.2 本地执行优化

数据比较小的话，可以使用本地模式

## 打开hive自动判断是否启动本地模式的开关
set hive.exec.mode.local.auto=true;
## map任务数最大值，不启用本地模式的task最大个数
set hive.exec.mode.local.auto.input.files.max=4;
## map输入文件最大大小，不启动本地模式的最大输入文件大小
set hive.exec.mode.local.auto.inputbytes.max=134217728;

1、只是 select * 的时候
2、where 条件针对分区字段进行筛选过滤时
3、带有 limit 分支语句时
不会执行mapreduce

3.3 JVM重用

MapReduce 中一个 MapTask 或者 ReduceTask 就会启动一个JVM 进程，一个 Task 执行完毕后，JVM 进程就会退出。设置多个

set mapred.job.reuse.jvm.num.tasks=5;

一个jvm运行5个task,串行执行，减少了jvm的开启。

3.4 并行执行

## 可以开启并发执行。
set hive.exec.parallel=true;
## 同一个sql允许最大并行度，默认为8。
set hive.exec.parallel.thread.number=16;

3.5 推测执行

负载不均衡或者资源分布不均等原因，会造成同一个作业的多个任务之间运行速度不一致。
Hadoop采用了推测执行（Speculative Execution）机制，为慢的任务启动一个备份任务，让该任务与原始任务同时处理同一份数据，并最终选用最先成功运行完成任务的计算结果作为最终结果。

# 启动mapper阶段的推测执行机制
set mapreduce.map.speculative=true;
# 启动reducer阶段的推测执行机制
set mapreduce.reduce.speculative=true;

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.


hive.mapred.reduce.tasks.speculative.execution
true
Whether speculative execution for reducers should be turned on.

3.6 严格模式

## 设置Hive的严格模式
set hive.mapred.mode=strict;
set hive.exec.dynamic.partition.mode=nostrict;

严格模式不允许执行

1、对于分区表，必须添加where对于分区字段的条件过滤
select * from student_ptn where age > 25
2、order by语句必须包含limit输出限制
select * from student order by age limit 100;
3、限制执行笛卡尔积的查询
select a.*, b.* from a, b;
4、在hive的动态分区模式下，如果为严格模式，则必须需要一个分区列式静态分区

4.调优实战 4.1 第一个例子：日志表和用户表做链接

users 表有 600w+ （假设有5G）的记录，把 users 分发到所有的 map 上也是个不小的开销，而且MapJoin 不支持这么大的小表。如果用普通的 join，又会碰到数据倾斜的问题。

--1.给users表做过滤，但是过滤的条件，先得到过滤的user_id
select distinct user_id from log;  -->a
--2.再获取日志信息的用户的信息
select b.* from a join users b on a.user_id=b.user_id;
--3.让c 和log 表进行做mapjoin --c小表
select c*,d* from c join log d on c.user_id=d.user_id;
和三位一
select  * from log a
left outer join (
select  d.*
from ( select distinct user_id from log ) c join users d on c.user_id =
d.user_id
) x
on a.user_id = x.user_id;

4.2 第二个例子：位图法求连续七天发朋友圈的用户

每天都要求 微信朋友圈 过去连续7天都发了朋友圈的小伙伴有哪些？
假设每个用户每发一次朋友圈都记录了一条日志。每一条朋友圈包含的内容：
日期，用户ID，朋友圈内容…
dt, userid, content, …
如果 微信朋友圈的 日志数据，按照日期做了分区
2020-07-06 file1.log(可能会非常大)
2020-07-05 file2.log
…
// 昨天和今天
select a.userid from table a join table b on a.userid = b.userid;
// 上一次join的结果 和 前天 join
…
// 上一次join的结构 和 大前天 join
…
假设微信有10E用户，我们每天生成一个长度为10E的二进制数组，每个位置要么是0，要么是1，如果为1，代
表该用户当天发了朋友圈。如果为0，代表没有发朋友圈。
然后每天：10E / 8 / 1024 / 1024 = 119M左右
求Join实现：两个数组做 求且、求或、异或、求反、求新增 bitmap