Hive 优化

1. 架构优化

　　Hive支持多种执行引擎，分别是 MapReduce、Tez、Spark、Flink。可以通过hivesite.xml文件中的hive.execution.engine属性控制。

矢量化查询执行：

　　矢量化查询(要求执行引擎为Tez)执行通过一次批量执行1024行而不是每行一行来提 高扫描，

聚合，过滤器和连接等操作的性能，这个功能一显着缩短查询执行时间。

set hive.vectorized.execution.enabled = true;
-- 默认 false
set hive.vectorized.execution.reduce.enabled = true;
-- 默认 false

备注：要使用矢量化查询执行，必须用ORC格式存储数据

成本优化器：

　　Hive的CBO是基于apache Calcite的，Hive的CBO通过查询成本(有analyze收集的统 计信息)会生成有效率的执行计划，

最终会减少执行的时间和资源的利用，使用CBO 的配置如下：

SET hive.cbo.enable=true; --从 v0.14.0默认
true
SET hive.compute.query.using.stats=true; -- 默认false
SET hive.stats.fetch.column.stats=true; -- 默认false
SET hive.stats.fetch.partition.stats=true; -- 默认true

　　定期执行表（analyze）的分析，分析后的数据放在元数据库中。

分区表：

　　对于一张比较大的表，将其设计成分区表可以提升查询的性能，对于一个特定分区的 查询，

只会加载对应分区路径的文件数据，所以执行速度会比较快。

　　分区字段的选择是影响查询性能的重要因素，尽量避免层级较深的分区，这样会造成 太多的子文件夹。一些常见的分区字段可以是：

• 日期或时间。如year、month、day或者hour，当表中存在时间或者日期字段时
• 地理位置。如国家、省份、城市等
• 业务逻辑。如部门、销售区域、客户等等

分桶表：

　　与分区表类似，分桶表的组织方式是将HDFS上的文件分割成多个文件。

　　分桶可以加快数据采样，也可以提升join的性能(join的字段是分桶字段)，因为分桶可 以确保某个key对应的数据在一个特定的桶内(文件)，

巧妙地选择分桶字段可以大幅 度提升join的性能。

　　通常情况下，分桶字段可以选择经常用在过滤操作或者join操作的字段。

存储格式：

　　存储格式一般需要根据业务进行选择，生产环境中绝大多数表都采用TextFile、 ORC、Parquet存储格式之一。

　　TextFile是最简单的存储格式，它是纯文本记录，也是Hive的默认格式。其磁盘开销 大，查询效率低，更多的是作为跳板来使用。

RCFile、ORC、Parquet等格式的表都 不能由文件直接导入数据，必须由TextFile来做中转。

　　Parquet和ORC都是Apache旗下的开源列式存储格式。列式存储比起传统的行式存 储更适合批量OLAP查询，并且也支持更好的压缩和编码。

选择Parquet的原因主要 是它支持Impala查询引擎，并且对update、delete和事务性操作需求很低。

压缩：

　　压缩技术可以减少map与reduce之间的数据传输，从而可以提升查询性能，关于压 缩的配置可以在hive的命令行中或者hive-site.xml文件中进行配置。

SET hive.exec.compress.intermediate=true

关于压缩的编码器可以通过mapred-site.xml, hive-site.xml进行配置，也可以通过命 令行进行配置，如：

-- 中间结果压缩
SET
hive.intermediate.compression.codec=org.apache.hadoop.io.compr
ess.SnappyCodec ;
-- 输出结果压缩
SET hive.exec.compress.output=true;
SET mapreduce.output.fileoutputformat.compress.codec =
org.apache.hadoop.io.compress.SnappyCodc

2.参数优化：

本地模式：

　　当Hive处理的数据量较小时，使用本地模式速度更快。

SET hive.exec.mode.local.auto=true; -- 默认 false
SET hive.exec.mode.local.auto.inputbytes.max=50000000;
SET hive.exec.mode.local.auto.input.files.max=5; -- 默认 4

一个作业只要满足下面的条件，会启用本地模式

• 输入文件的大小小于 hive.exec.mode.local.auto.inputbytes.max 配置的大 小
• map任务的数量小于 hive.exec.mode.local.auto.input.files.max 配置的 大小
• reduce任务的数量是1或者0

严格模式：

　　所谓严格模式，就是强制不允许用户执行3种有风险的HiveQL语句。

• 查询分区表时不限定分区列的语句；
• 两表join产生了笛卡尔积的语句；
• 用order by来排序，但没有指定limit的语句。

要开启严格模式，需要将参数 hive.mapred.mode 设为strict(缺省值)。

该参数可以不在参数文件中定义，在执行SQL之前设置(set hive.mapred.mode=nostrict )

JVM重用：

  当执行轻量级作业，开启JVM重用会比较有效。

# 代表同一个MR job中顺序执行的5个task重复使用一个JVM，减少启动和关闭的开销
SET mapreduce.job.jvm.numtasks=5;

　　这个功能的缺点是，开启JVM重用将一直占用使用到的task插槽，以便进行重用，直 到任务完成后才能释放。

如果某个“不平衡的”job中有某几个reduce task执行的时间 要比其他Reduce task消耗的时间多的多的话，

那么保留的插槽就会一直空闲着却无 法被其他的job使用，直到所有的task都结束了才会释放。

 并行执行：

Hive的查询通常会被转换成一系列的stage，这些stage之间并不是一直相互依赖的， 可以并行执行这些stage，通过下面的方式进行配置：

SET hive.exec.parallel=true; -- 默认false
SET hive.exec.parallel.thread.number=16; -- 默认8

　　并行执行可以增加集群资源的利用率，如果集群的资源使用率已经很高了，那么并行 执行的效果不会很明显。

推测执行：

　　启动备份任务，和原来的任务同时执行，处理同一份数据，先完成的作为最终结果。

set mapreduce.map.speculative=true
set mapreduce.reduce.speculative=true
set hive.mapred.reduce.tasks.speculative.execution=true

小文件合并：

 在map执行前合并小文件，减少map数：

# 缺省参数
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

在Map-Reduce的任务结束时合并小文件：

# 在 map-only 任务结束时合并小文件，默认true
SET hive.merge.mapfiles = true;
# 在 map-reduce 任务结束时合并小文件，默认false
SET hive.merge.mapredfiles = true;
# 合并文件的大小，默认256M
SET hive.merge.size.per.task = 268435456;
# 当输出文件的平均大小小于该值时，启动一个独立的map-reduce任务进行文件
merge
SET hive.merge.smallfiles.avgsize = 16777216;

Fetch模式：

　　Fetch模式是指Hive中对某些情况的查询可以不必使用MapReduce计算。select col1, col2 from tab ;

可以简单地读取表对应的存储目录下的文件，然后输出查询结果到控制台。

在开启 fetch模式之后，在全局查找、字段查找、limit查找等都不启动 MapReduce 。

# Default Value: minimal in Hive 0.10.0 through 0.13.1, more
in Hive 0.14.0 and later
hive.fetch.task.conversion=more

3.SQL优化

列裁剪和分区裁剪：

　　列裁剪是在查询时只读取需要的列；分区裁剪就是只读取需要的分区。

　　简单的说：select 中不要有多余的列，坚决避免 select * from tab; 查询分区表，不读多余的数据；

sort by 代替 order by：

　　order by 是全局排序，只有一个reduce

　　如果使用sort by，那么还是会视情况启动多个reducer进行排序，并且保证每个 reducer内局部有序。

　　为了控制map端数据分配到reducer的key，往往还要配合 distribute by 一同使用。

　　如果不加 distribute by 的话，map端数据就会随机分配到 reducer。

group by 代替 count(distinct)：

　　当要统计某一列的去重数时，如果数据量很大，count(distinct) 会非常慢。

原因与 order by类似，count(distinct)逻辑只会有很少的reducer来处理。

　　

-- 原始SQL
select count(distinct uid)
from tab;
-- 优化后的SQL
select count(1)
from (select uid
from tab
group by uid) tmp;

即：在group by 外层统计数量

group by 配置调整：

 map 端预聚合：group by时，如果先起一个combiner在map端做部分预聚合，可以有效减少shuffle 数据量。

-- 默认为true
set hive.map.aggr = true
set hive.groupby.mapaggr.checkinterval = 100000

//通过 hive.groupby.mapaggr.checkinterval 参数也可以设置map端预聚合的行数 阈值，超过该值就会分拆job，默认值10W。

join 基础优化:

一般不特殊指定连接方式使用的都是common join，性能较差。

map join ：适用于大表和小表。

bucket map join：

　　分桶连接：Hive 建表的时候支持hash 分区通过指定clustered by (col_name,xxx ) into number_buckets buckets 关键字.

当连接的两个表的join key 就是bucket column 的时候，就可以通过设置hive.optimize.bucketmapjoin= true 来执行优 化。

　　原理：通过两个表分桶在执行连接时会将小表的每个分桶映射成hash表，每个task 节点都需要这个小表的所有hash表，

但是在执行时只需要加载该task所持有大表分 桶对应的小表部分的hash表就可以，所以对内存的要求是能够加载小表中最大的 hash块即可。

　　注意点：小表与大表的分桶数量需要是倍数关系，这个是因为分桶策略决定的，分桶 时会根据分桶字段对桶数取余后决定哪个桶的，所以要保证成倍数关系。

处理空值或无意义值：

　　空值字段做表连接很慢。可以将空值key用随机方式打散，例如将用户ID为null的记录随机改为负值：

select a.uid, a.event_type, b.nickname, b.age
from (
select
(case when uid is null then cast(rand()*-10240 as int) else
uid end) as uid,
event_type from calendar_record_log
where pt_date >= 20190201
) a left outer join (
select uid,nickname,age from user_info where status = 4
) b on a.uid = b.uid;

单独处理倾斜key：

　　把倾斜key放入临时表，打上一个较小的随机数前缀（比如 0~9），最后再进行聚合。

　　不要一个Select语句中，写太多的Join。一定要了解业务，了解数据。

(A0-A9) 分成多条语句，分步执行；(A0-A4; A5-A9)；先执行大表与小表的关联；

合理设置maptask ，reducetask的数量。

设置maptask 和reducetask能处理的数据量 256-》512M，这样可以减少maptask，reducetask 数量

减少stage数量，例如多个insert from 字句提前