Hive优化整理

压缩格式

​ 常见的压缩格式有Sequence File、RC、ORC、Parquet。生产环境中hive一般用orc，spark一般用parquet，snappy配合parquet性能最高。

注意点

1）orc格式的表不能通过load data导入数据

2）Sequence 可分割，与hadoop文件兼容

3）ORC快速存取，列压缩

Hive函数

内置函数

1.NVL( value，default_value) ，给值为 NULL 的数据赋值
2.CASE WHEN THEN ELSE END
CASE a WHEN b THEN c [WHEN d THEN e]* [ELSE f] END
说明：如果 a 等于 b，那么返回 c；如果 a 等于 d，那么返回 e；否则返回 f
3.CONCAT(string A/col, string B/col…)：返回输入字符串连接后的结果，支持任意个输入
字符串;
4.CONCAT_WS(separator, str1, str2,…)：它是一个特殊形式的 CONCAT()。第一个参数剩
	余参数间的分隔符。分隔符可以是与剩余参数一样的字符串。如果分隔符是 NULL，返
	回值也将为 NULL。这个函数会跳过分隔符参数后的任何 NULL 和空字符串。分隔符
	将被加到被连接的字符串之间;
5.COLLECT_SET(col)：函数只接受基本数据类型，它的主要作用是将某字段的值进行去
	重汇总，产生 array 类型字段。
6.EXPLODE(col)：将 hive 一列中复杂的 array 或者 map 结构拆分成多行。
7.LATERAL VIEW
用法：LATERAL VIEW udtf(expression) tableAlias AS columnAlias
解释：用于和 split, explode 等 UDTF 一起使用，它能够将一列数据拆成多行数据，在此
	基础上可以对拆分后的数据进行聚合。
8.OVER()：指定分析函数工作的数据窗口大小，这个数据窗口大小可能会随着行的变而
变化。
	CURRENT ROW：当前行
	n PRECEDING：往前 n 行数据
	n FOLLOWING：往后 n 行数据
	UNBOUNDED：起点，UNBOUNDED PRECEDING 表示从前面的起点， UNBOUNDED
	FOLLOWING 表示到后面的终点
	LAG(col,n,default_val)：往前第 n 行数据
	LEAD(col,n, default_val)：往后第 n 行数据

9.RANK() 排序相同时会重复，总数不会变
10.DENSE_RANK() 排序相同时会重复，总数会减少
11.ROW_NUMBER() 会根据顺序计算
12.NTILE(n)：把有序分区中的行分发到指定数据的组中，各个组有编号，编号从 1 开始，
对于每一行，NTILE 返回此行所属的组的编号。注意：n 必须为 int 类型。用于将分组
数据按照顺序切分成 n 片，返回当前切片值，如果切片不均匀，默认增加第一个切片的
分布。

自定义函数

• UDF（User-Defined-Function）一进一出
• UDAF（User-Defined Aggregation Funcation）聚集函数，多进一出
• UDTF（User-Defined Table-Generating Functions）一进多出，如lateral view explore()

Hive优化

1.合理设置map reduce的task数量

1.1 map阶段优化

​ 直接调整mapred.map.tasks这个参数是没有效果的。需要改变mapred.max/min.split.size来改变task数量。并且，map数量需要我们衡量map任务启动和初始化的时间与逻辑处理的时间以达到资源合理利用。

相关参数：

set mapred.max.split.size=100000000;
set mapred.min.split.size.per.node=100000000;
set mapred.min.split.size.per.rack=100000000;
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; //小文件合并

1.2 reduce阶段优化

​ 如果reduce太小可能发生OOM，太大会产生很多小文件，影响namenode。

​ 如果不指定mapred.reduce.tasks/mapreduce.job.reduces，hive会根据输入文件大侠奥估算reduce数量，这种方式并不准确。

1）hive计算reduce的公式为：N=min（hive.exec.reducers.max 每个任务最大reduce数，总输入数据量/hive.exec.reducers.bytes.per.reducer 每个reduce处理的数据量）

2）可以调整hive.exec.reducers.bytes.per.reducer或mapred.reduce.tasks来改变reduce个数

3）出现一个reduce的原因有：没有使用group by 、用了order by、有笛卡尔积

2.小文件合并优化

2.1 合并参数

是否合并Map输出文件：hive.merge.mapfiles=true（默认值为true）
是否合并Reduce端输出文件：hive.merge.mapredfiles=false（默认值为false）
合并文件的大小：hive.merge.size.per.task=256*1000*1000（默认值为256000000）

2.2 小文件的产生

• 动态分区插入数据，产生大量的小文件，从而导致map数量剧增；
• reduce数量越多，小文件也越多（reduce的个数和输出文件是对应的）；
• 数据源本身就包含大量的小文件。

2.3 解决方案

• 从源头解决
  • 不用textfile存储格式，可一定程度避免小文件
  • 减少reduce数量
  • 少用动态分区，用时记得按distribute by分区
• 已有文件
  • 使用hadoop archive命令把小文件进行归档
  • 重建表，建表时减少reduce数量
  • 调整参数

//每个Map最大输入大小(这个值决定了合并后文件的数量)  
set mapred.max.split.size=256000000;    
//一个节点上split的至少的大小(这个值决定了多个DataNode上的文件是否需要合并)  
set mapred.min.split.size.per.node=100000000;  
//一个交换机下split的至少的大小(这个值决定了多个交换机上的文件是否需要合并)    
set mapred.min.split.size.per.rack=100000000;  
//执行Map前进行小文件合并  
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;   

设置map输出和reduce输出进行合并的相关参数：
[java] view plain copy
//设置map端输出进行合并，默认为true  
set hive.merge.mapfiles = true  
//设置reduce端输出进行合并，默认为false  
set hive.merge.mapredfiles = true  
//设置合并文件的大小  
set hive.merge.size.per.task = 256*1000*1000  
//当输出文件的平均大小小于该值时，启动一个独立的MapReduce任务进行文件merge。  
set hive.merge.smallfiles.avgsize=16000000

SQL优化

1.列裁剪

只查询需要的列，避免读取无关列。相关参数为hive.optimize.cp=true（默认值为真）

2.分区裁剪

查询数据时指定分区，避免全表扫描。相关参数为hive.optimize.pruner=true（默认值为真）

3.熟练使用SQL

写出高效率的SQL语句。例如，少用临时表，节省临时表的读写开销

4.不同数据类型关联导致的数据倾斜

避免关联字段的数据类型不同，例如，string和int关联会导致数据都分配到一个reduce。

5.利用UNION ALL的特性

多表union all会优化成一个job，如果同一个表多次用到只会扫描一次。需要注意的是，union all的优化不适用于嵌套查询。

6.COUNT(DISTINCT)优化

数据量大的情况下只会起一个reduce，需要用group by和count的方式替换

7.JOIN优化

7.1大小表join

​ 新版本不需要将小表放在左边，hive会自动优化。

7.2两个大表Join

1）过滤空key

2）空key转换

8.Group BY

当一个key过大时会发生数据倾斜，可以先进行map join。还可以开启负债均衡通过hive.groupby.skewindata参数。

9.优化in/exists语句

虽然hive支持in/exists，但还是推荐使用hive的一个高效替代方案：left semi join

10.排序

• cluster by: 对同一字段分桶并排序。cluster by 除了具有 distribute by 的功能外还兼具 sort by 的功能。但是排序只能是升序排序。
• distribute by + sort by: 分桶，保证同一字段值只存在一个结果文件当中，结合sort by 保证每个reduceTask结果有序；
• sort by: 单机排序，单个reduce结果有序
• order by：全局排序，缺陷是只能使用一个reduce

11.压缩格式

Hive支持ORCfile，这是一种新的表格存储格式，可以通过诸如谓词下推，压缩等技术来提高执行速度提升。

参考参数

set mapreduce.output.fileoutputformat.compress=true // 默认值是 false
set mapreduce.output.fileoutputformat.compress.type=BLOCK // 默认值是 Record
set mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.GzipCodec // 默认值是 org.apache.hadoop.io.compress.DefaultCodec
set mapred.map.output.compress=true
set mapreduce.map.output.compress.codec=org.apache.hadoop.io.compress.GzipCodec // 默认值是 org.apache.hadoop.io.compress.DefaultCodec 
set hive.exec.compress.output=true // 默认值是 false，不压缩
set hive.exec.compress.intermediate=true // 默认值是 false，为 true 时 MR 设置的压缩才启用

12.引擎的选择

​ Hive可以使用Apache Tez执行引擎而不是古老的Map-Reduce引擎。没有在环境中没有默认打开，在Hive查询开头将以下内容设置为‘true’来使用Tez：“设置hive.execution.engine = tez; ”，通过上述设置，你执行的每个HIVE查询都将利用Tez。

13.使用向量化查询

向量化查询执行通过一次性批量执行1024行而不是每次单行执行，从而提供扫描、聚合、筛选器和连接等操作的性能。在Hive 0.13中引入，此功能显着提高了查询执行时间，并可通过两个参数设置轻松启用：

设置hive.vectorized.execution.enabled = true;
　　设置hive.vectorized.execution.reduce.enabled = true;

14.cost based query optimization

​ Hive自0.14.0开始，加入了一项“Cost based Optimizer”来对HQL执行计划进行优化，这个功能通过“hive.cbo.enable”来开启。在Hive 1.1.0之后，这个feature是默认开启的，它可以自动优化HQL中多个JOIN的顺序，并选择合适的JOIN算法。

参考参数

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

15.模式选择

• 本地模式

  对于少量的数据可以让其在单台机器上完成，效率更高。

set hive.exec.mode.local.auto=true;  //开启本地mr
//设置local mr的最大输入数据量，当输入数据量小于这个值时采用local  mr的方式，默认为134217728，即128M
set hive.exec.mode.local.auto.inputbytes.max=50000000;
//设置local mr的最大输入文件个数，当输入文件个数小于这个值时采用local mr的方式，默认为4
set hive.exec.mode.local.auto.input.files.max=10;

• 并行模式

  Hive会将一个查询转化成一个或多个阶段。这样的阶段可以是MapReduce阶段、抽样阶段、合并阶段、limit阶段。不存在依赖的阶段可以并行。可以设置set hive.exec.parallel=true

• 严格模式 （set Hive.mapred.modestrict)

  1）分区表必须指定分区

  2）order by必须使用limit

  3）不允许笛卡尔积查询

16.JVM重用

如果map、reduce数量过多，jvm的启动会造成相当大的开销。可以开启JVM重用mapred.job.reuse.jvm.num.tasks

17.推测执行

对于比较慢的任务再开启一个相同的任务，哪个先执行完则提交。可以设置参数：

mapred.map.tasks.speculative.execution=true
mapred.reduce.tasks.speculative.execution=true