Hive的一般学习者和培训者在谈性能优化的时候一般都会从语法和参数这些雕虫小技的角度谈优化,而不会革命性的优化Hive的性能,产生这种现象的原因有:1,历史原因和思维定势:大家学习SQL的时候一般都是就单机DB,这个时候你的性能优化技巧确实主要是SQL语法和参数调优;2,Hive的核心的性能问题往往是产生在超过规模数据集,例如说100亿条级别的数据集,以及每天处理上千上万个Hive作业的情况下产生的;上面的第二点是我们现在Hive性能调优部分要彻底解决的内容;要从根本上解决和显著的解决实际企业中Hive真正的性能优化问题,必须考虑到底什么是Hive性能的限制,我们按照优先级来说:第一重要的是:战略性架构 解决海量数据下大量Job过于频繁的IO问题,而这个问题实质上涉及了架构方面的分表 数据复用 以及分区表等调优的方式; 补充:1,海量的数据中有些数据是高频使用的数据,而有些是很少使用的,如果能够分离成为不同的表,会极大的提升效率;很多的作业可能会有共同点,抽离出来先进行计算并保留计算结果,后面的作业都可以复用;同时,底层的基础功能也可以先计算,在上层应用的时候直接拿数据结果,而不是每次都重复计算; 2,合理从用静态分区表和动态分区表,可以避免数据全局扫描及计算资源更合理的利用; 3,数据倾斜的一站式解决方案;第二重要的是:引擎和物理层面,很多内容都是普通Hive使用这不知道的! 从Hive语法和Job内部的角度去进行优化,这要求MapReduce以及Hive如何被翻译成为MapReduce要非常精通;第三重要的是:一些关键的参数;归根到底,Hive的性能优化主要考虑的是如何最大化和最有效的使用CPU Memory IO;
Hive背后的Mapper调优:
1,Mapper数过大,会产生大量小文件,由于Mapper是基于虚拟机的,过多的Mapper创建和初始化及关闭虚拟机都会消耗大量的硬件资源;
Mapper数太小,并发度过小,Job执行时间过长,无法充分利用分布式硬件资源;
2,Mapper数据由什么决定呢?
输入文件数目;
输入文件的大小;
配置参数;
默认情况下:例如一个文件800M,BLock大小是128M,那么Mapper数目就是7个,6个Mapper处理的数据是 128M, 1个Mapper处理的数据是32M;再例如,一个目录下有三个文件分别大小问5M 10M 150M
此时会产生4个Mapper,处理的数据分别是5M 10M 128M 22M;
减少Mapper的个数,就要合并小文件,这种小文件有可能是直接来自于数据源的小文件,也可能是Reducer产生的小文件;
set hive.input.format=org.apache.Hadoop.hive.ql.io.CombineHiveInputFormat;
set hive.merge.mapFiles=true;
set hive.merge.mapredFiles=true;
set hive.merge.size.per.task=256000000
set mapred.max.split.size=256000000
set mapred.min.split.size.per.node=128000000
增加Mapper的个数,一般是通过控制Hive SQL中上一个Job的Reducer个数来控制的,例如在Join操作的时候会把多个表分解为多个Job;
set mapred.map.tasks=2;
set hive.merge.mapFiles=true;
set hive.merge.mapredFiles=true;
set hive.merge.size.per.task=256000000
例如我们有5个300M的文件;按照上面的配置会产生10个Mapper,5个Mapper处理的都是256M的数据,另外5个Mapper处理的都是44M的数据,问题是:大的Mapper会数据倾斜
如何解决,设置set mapred.map.tasks=6,此时根据MapRed的运行机制,会划分6个Mapper,每个Mapper的处理数据的大小是250M, min(1500M/6, 256M) =250M
Hive背后的Reducer调优:
1,Reducer数目过大的话,会产生很多小文件,每个Reducer都会产生一个文件,如果这些小文件是下一个JOB的输入,则会需要对小文件进行合并;同样启动 初始化和销毁Reducer的虚拟机也需要消耗大量的硬件;
Reducer数据过小的话,Reduce的时间会比较长,也可能会出现数据倾斜;
2,如何控制Reducer的个数呢?
set hive.exec.reducers.byte.per.reducer=1G
set hive.exec.reducers.max=999
Reducer个数=min(999, Reducer的数据输入总量/1G);
set mapred.reduce.tasks = 10, 默认是1; 如果说当前的Reducer的结果很大,且被接下来多个Job使用其结果,我们该如何设置参数呢?一般都需要调大该参数;
什么情况下只有一个Reducer?如果不进行Group by但却需要汇总,或者说Order by,当然如果最后Reducer的数据小于默认的1G的话,也会只有一个Reducer;
1,Hive在分布式运行的时候最害怕的是数据倾斜,这是由于分布式系统的特性决定的,因为分布式系统之所以很快是由于作业平均分配给了不同的节点,不同节点同心协力,从而达到更快处理完作业的目的;
顺便说明一下,处理数据倾斜的能力是hadoop和Spark工程师最核心的竞争力之一;
2,Hive中数据倾斜的原因:
数据在分布式节点上分布不平衡;
join时某些key可能特别大;
groupBy的时候某个Key可能特别多;
count(distinct)有可能出现数据倾斜,因为其内部首先会进行groupBy操作;
3,join,我们希望join时候key是分散,如果一个key的数据量特别大,有可能会出现数据倾斜和OOM,一个核心点是:小表join大表,在reduce阶段左侧的小表会加载进内存,减少OOM的风险;
4,大表join大表的情况:数据倾斜,例如null值,解决办法一般是要打散null值,例如说使用随机数等,如果数据倾斜比较严重,采用这种方式可以提升至少一倍的速度;
5,mapJoin:小表join(超)大表的时候,可以采用mapJoin的方式把小表全部加载到Mapper端的内存中/*+MAPJOIN(table_name)*/;
6,小表join(超)大表的时候,是否会自动进行mapJoin,想进行mapJoin,需要设置:set hive.auto.convert.join=true,Hive在进行join的时候会判断左表的大小来决定是否进行mapJoin:
set hive.mapjoin.smalltable.filesize=128000000;
set hive.mapjoin.cache.numrows=100000;
上述参数可以根据实际的硬件机器的内存进行调整,对性能有至关重要的影响,因为没有了Shuffle;
对于mapJoin我们能够使用Mapper端JVM中多大的内存呢?
set hive.mapjoin.followby.gby.localtask.max.momery.usage = 0.8
set hive.mapjoin.localtask.max.memory.uage=0.9
7,groupBy,我们可以设置在Mapper端进行部分聚合,最后在Reducer端进行全局聚合
set hive.map.aggr=true;
set hive.groupby.mapaggr.checkinterval=100000
set hive.groupby.skewindata = true 内部会产生两个Job,第一个Job会通过自己的算法打散倾斜的Key并进行聚合操作且保留结果,第二个Job会完成全部的groupBy操作,会产生Mapper-Reducer-Reducer的结构
8, count(distinct),如果某个字段特别多,容易产生数据倾斜,解决思路:
在查询语句中例如对null进行过滤,在结果中加1
9, 笛卡尔积:join时候没有on条件,或者on条件无效,这个时候会使用Reducer进行笛卡尔积的操作;