【Hive笔记】五、调优指南

概述

如下很多配置其实在 hive 的新版本已经优化啦, 通用配置放置 $HOME/.hiverc 即可持续生效

各种场景的优化有所不同, 可以利用 analyze table 进行分析.

通用优化

表连接优化

数据量大的表放在后面 Hive假定查询中最后的一个表是大表.它会将其它表缓存起来,然后扫描最后那个表.因此通常需要将小表放前面,或者标记哪张表是大表:、

利用 set hive.auto.convert.join = true;也可生效
```
SELECT
/*+ STREAMTABLE(a) */
a.val, b.val, c.val
FROM a
JOIN b
ON (a.key = b.key1)
JOIN c
ON (c.key = b.key1)
```
使用相同的连接键

当对3个或者更多个表进行join连接时,如果每个on子句都使用相同的连接键的话,那么只会产生一个MapReduce job.
尽量尽早地过滤数据

减少每个阶段的数据量,对于分区表要加分区,同时只选择需要使用到的字段.
尽量原子化操作

尽量避免一个SQL包含复杂逻辑,可以使用中间表来完成复杂的逻辑

用insert into替换union all

order by & sort by

order by : 对查询结果进行全局排序消耗时间长,需要set hive.mapred.mode=nostrict
sort by : 局部排序,并非全局有序,提高效率.

成本优化器

传统的数据库,成本优化器做出最优化的执行计划是依据统计信息来计算的.

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 ;

场景优化

limit 语句快速出结果

一般情况下,Limit语句还是需要执行整个查询语句,然后再返回部分结果.有一个配置属性可以开启,避免这种情况—对数据源进行抽样.

缺点:有可能部分数据永远不会被处理到

hive.limit.optimize.enable=true --- 开启对数据源进行采样的功能
hive.limit.row.max.size --- 设置最小的采样容量
hive.limit.optimize.limit.file --- 设置最大的采样样本数

本地模式

对于小数据集,为查询触发执行任务消耗的时间>实际执行job的时间,因此可以通过本地模式,在单台机器上(或某些时候在单个进程上)处理所有的任务.

set oldjobtracker=${hiveconf:mapred.job.tracker};
set mapred.job.tracker=local; 　
set marped.tmp.dir=/home/edward/tmp;
set mapred.job.tracker=${oldjobtracker};
sql 语句

可以通过设置属性hive.exec.mode.local.auto的值为true,来让Hive在适当的时候自动启动这个优化,也可以将这个配置写在$HOME/.hiverc文件中. 当一个job满足如下条件才能真正使用本地模式:

job的输入数据大小必须小于参数:hive.exec.mode.local.auto.inputbytes.max(默认128MB)
job的map数必须小于参数:hive.exec.mode.local.auto.tasks.max(默认4)
job的reduce数必须为0或者1
可用参数hive.mapred.local.mem(默认0)控制child的jvm使用的最大内存数.

并行执行

Hive会将一个查询转化为一个或多个阶段,包括:MapReduce阶段、抽样阶段、合并阶段、limit阶段等.默认情况下,一次只执行一个阶段. 不过,如果某些阶段不是互相依赖,是可以并行执行的.

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

数据倾斜

表现:任务进度长时间维持在99%(或100%),查看任务监控页面,发现只有少量(1个或几个)reduce子任务未完成.因为其处理的数据量和其他reduce差异过大.单一reduce的记录数与平均记录数差异过大,通常可能达到3倍甚至更多. 最长时长远大于平均时长.

原因:

key分布不均匀
业务数据本身的特性
建表时考虑不周
某些SQL语句本身就有数据倾斜

hive.map.aggr = true
hive.groupby.skewindata=true

注意: 上述参数只是一种方式, 具体问题具体分析,查看异常聚集的数据是否为无效数据,是否可以剔除如果不能,利用加前缀的方式进行手动分组,再聚合.

Each-Top-N 优化

执行类似如下操作的时候

SELECT *,Rank()Over(Partition BY)
FROM tblTable
Distribute By FieldA
Sort By FieldA, MetricA DESC

可采用 predicate pushdown (PPD) 在存储级别执行过滤

该部分含义为: 过滤条件在map端执行,减少了map端的输出,降低了数据在集群上传输的量,节约了集群的资源,也提升了任务的性能.

SET hive.optimize.ppd=true ;
SET hive.optimize.ppd.storage=true ;

配置参数

调整mapper和reducer的个数

Map阶段优化

map个数的主要的决定因素有

input的文件总个数,input的文件大小
集群设置的文件块大小(默认128M,不可自定义).

参考举例如下:

假设input目录下有1个文件a,大小为780M,那么hadoop会将该文件a分隔成7个块(6个128m的块和1个12m的块),从而产生7个map数

假设input目录下有3个文件a,b,c,大小分别为10m,20m,130m,那么hadoop会分隔成4个块(10m,20m,128m,2m),从而产生4个map数.

即如果文件大于块大小(128m),那么会拆分,如果小于块大小,则把该文件当成一个块.

map执行时间:map任务启动和初始化的时间+逻辑处理的时间.

减少map数

若有大量小文件(小于128M),会产生多个map,处理方法是:
```
set mapred.max.split.size=100000000;
set mapred.min.split.size.per.node=100000000;
set mapred.min.split.size.per.rack=100000000;
```
前面三个参数确定合并文件块的大小,大于文件块大小128m的,按照128m来分隔,小于128m,大于100m的,按照100m来分隔,把那些小于100m的(包括小文件和分隔大文件剩下的)进行合并.
```
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; -- 执行前进行小文件合并.
```
增加map数

当input的文件都很大,任务逻辑复杂,map执行非常慢的时候,可以考虑增加Map数,来使得每个map处理的数据量减少,从而提高任务的执行效率.
```
set mapred.map.tasks=?
```

Reduce阶段优化

set mapred.reduce.tasks=?
set hive.exec.reducers.bytes.per.reducer = ?

一般根据输入文件的总大小,用它的estimation函数来自动计算reduce的个数:reduce个数 = InputFileSize / bytes per reducer

压缩设置(未使用过)

--  map/reduce 输出压缩(一般采用序列化文件存储)
set hive.exec.compress.output=true;
set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;
set mapred.output.compression.type=BLOCK;
--任务中间压缩
set hive.exec.compress.intermediate=true;
set hive.intermediate.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;
set hive.intermediate.compression.type=BLOCK;

动态分区(插入多个分区数据)

可以通过动态分区来进行多个分区的插入, 但是只可以一个分区是动态的,还有一点比较坑,就是如果新增列,记得动态的分区字段放在 select 的最后一个

--设置非严格模式
set hive.exec.dynamic.partition.mode=nonstrict;
--设置动态分区
set hive.exec.dynamic.partition=true;
--设置动态分区每个节点最多可划分为多少个分区
set hive.exec.max.dynamic.partitions.pernode=1000;
--设置动态分区时的分区最大数量
set hive.exec.max.dynamic.partitions=2000;

严格模式

set hive.marped.mode=strict --防止用户执行那些可能意想不到的不好的影响的查询

分区表,必须选定分区范围
对于使用order by的查询,要求必须使用limit语句.因为order by为了执行排序过程会将所有的结果数据分发到同一个reducer中进行处理
限制笛卡尔积查询:两张表join时必须有on语句

transform+python

一种嵌入在hive取数流程中的自定义函数,通过transform语句可以把在hive中不方便实现的功能在python中实现,然后写入hive表中.示例语法如下:

select transform({column names1})
using '**.py'
as {column names2}
from {table name}

如果除python脚本外还有其它依赖资源,可以使用ADD ARVHIVE.

参考文献

大数据框架

#Hive

【Hive笔记】五、调优指南

https://www.windism.cn/2012667041.html

作者

windism

发布于

2021年2月20日

许可协议

【风控模型】特征工程之类别不平衡问题上一篇

【读书笔记】结构性改革-黄奇帆下一篇

【Hive笔记】 五、调优指南

概述

通用优化

表连接优化

用insert into替换union all

order by & sort by

成本优化器

场景优化

limit 语句快速出结果

本地模式

并行执行

数据倾斜

Each-Top-N 优化

配置参数

调整mapper和reducer的个数

Map阶段优化

Reduce阶段优化

压缩设置(未使用过)

动态分区(插入多个分区数据)

严格模式

transform+python

参考文献

【Hive笔记】五、调优指南