hive优化分享

Posted sunhuaj

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了hive优化分享相关的知识,希望对你有一定的参考价值。

粘贴一下我在部门中的一次hive优化的分享。

 

简述

hive构建在hadoop基础上,利用分布式存储,通过mr引擎实现对大数据的计算。MR会频繁地读写磁盘而且MR任务的启动成本很高。对于hive优化显得尤为重要。而优化的核心就是更好地利用hadoop的分布式特性和hive的有点。本篇从IO、参数设置、案例实战来说明如何优化我们的hive。受限于个人能力,如有不足之处,还望指出,一起沟通讨论。

 

1、IO

  A、通过列裁剪,只读取需要的列[对select * 的做法应进行严格要求,甚至禁止]

  B、 join操作前,应提前处理表,只读取需要的列,并对数据进行梳理。

      原:select A.char1,B.char2

           from A join B

                 on A.id=B.id

           where A.disabled=0 and B.disabled=0

      优:select A.char1,B.char2

            from

                 (

                  select id,char1

                    from a

                   where a.disabled=0

                 ) A

                 join

                 (

                  select id,char2

                    from b

                   where b.disabled=0

                 ) B

                 on A.id=B.id

  C、选择合适的存储类型

    建议使用orc存储格式,作为默认设置[支持切片,有压缩,提高查询速度]。创建表时,指定stored as orcfile

  D、走分区[可选]

    减少读取更多的分区文件,极大地有帮助!处理分区时,不可对分区字段进行操作,否则将会导致全表扫描。

2、join

    A、left semi join取代in

      根据测试,对于在大数据量中查询某些取值时,有效提升查询效率。[测试:从erp_fct_order表中,查找若干数据。left semi join 使用了100‘+,in耗时300‘+]

    B、map join

      当小表与大表进行关联时,通过map join,使左表进入内存,与右表进行关联。这个阶段不涉及reduce。

      通过设置set hive.auto.convert.join=true;[默认]

              set hive.mapjoin.smalltable.filesize=2500000;[25M,通过配置该属性来确定使用该优化的表的大小,如果表的大小小于此值就会被加载进内存中]

3、设置执行参数

  reducer数量

      我们执行语句时,总会看到这个: 

      Number of reduce tasks determined at compile time: 1

      In order to change the average load for a reducer (in bytes):

        set hive.exec.reducers.bytes.per.reducer=<number>

      In order to limit the maximum number of reducers:

        set hive.exec.reducers.max=<number>

      In order to set a constant number of reducers:

        set mapreduce.job.reduces=<number>

      Starting Job = job_1515122600431_869586, Tracking URL = http://hadoop04:8088/proxy/application_1515122600431_869586/

 

      上面其实就是,hadoop确定reduce的数量的方式。

      N=min(hive.exec.reducers.max,总数据量大小/hive.exec.reducers.bytes.per.reducer),默认情况下,hive.exec.reducers.max=999,hive.exec.reducers.bytes.per.reducer=1G;

      我们也可以强制指定reduce数量,mapreduce.job.reduces=N。

      需要注意的是,有的reduce为0,有的只为1。

      为0,即为该任务只有map,没有reduce。

      为1,即为全局,如order by、count、笛卡尔积。

  mapper的数量

      map端处理文件时,将对文件进行拆分成若干个task数。而如何拆分,这本身和文件的存储类型相关。hive系统默认的类型为org.apache.hadoop.hive.ql.io.CombineHiveInputFormat[set hive.input.format;],当我们不指定存储类型时,hive将根据文件本身设置对应的存储类型。如:org.apache.hadoop.mapred.TextInputFormat。

      继承了FileInputFormat的切分算法:splitSize = max{split.size,min{goalSize,blockSize}}

      若格式为:HiveInputFormat

      split.size:mapred.min.split.size,默认为1。切片最小大小,没有什么能比这个更小的了。

      goalSize:目标大小,为用户设置的map task数与实际大小决定。goalSize=总大小/设置的mapred.map.tasks的大小[默认为2]

      blockSize:块大小[set dfs.block.size],当前我们系统的块为:128M

      所以,当我们没有设置最小的切片大小时,切片的大小,取决于文件大小、设置的mapred.map.tasks数、块大小.

      例如:一个文件如果是300M,则切片大小为:splitSize=max{1,min{300/2,128}}=128M,则3个map对应的文件大小为:128,128,46

      若格式为:TextInputFormat:

      split.size:mapred.max.split.size,[当前系统为256M],则splitSize为max{256,min{300/2,128}}=256,则只有一个map.

    小结

  mapper的数量合适,单个mapper处理的数据量合适;拆分大小尽量与block数据块相同,避免一个拆分块大小有多个 hdfs 块,且位于不同数据节点,从而降低网络 IO。如果想减少mapper的数据量,一般只要将mapred.max.split.size设置的更大就可以了。

  合并小文件

  hive基于NameNode进行调度管理和任务分发,NameNode本身是有瓶颈的,有固定的大小。在hadoop任务中,NameNode加载必要的信息,其中包括各个文件的元数据(位置、大小、块信息等)。一般情况下,当文件数达到1000W时,所占用的NameNode内存达到3G,将带来性能下降。Hadoop本身对大文件具有很高的处理能力,但在mr任务过程中,map任务和reduce任务都会带来小文件的问题。当小文件作为输入,由map来进行处理时,启用map带来的时间比处理更耗时。此外,reduce数量多的话,产生的小文件也会增多。

  输入时合并

  set mapred.max.split.size=256000000;-- 每个Map最大输入大小,决定合并后的文件数      

  set mapred.min.split.size.per.node=100000000; -- 一个节点上split的至少的大小 ,决定了多个data node上的文件是否需要合并 [当前设置为:1,默认设置。设置该参数,一定要设置下方的rack参数,否则报错。]

  set mapred.min.split.size.per.rack=100000000; -- 一个交换机下split的至少的大小,决定了多个交换机上的文件是否需要合并 [当前设置为:1,默认设置]

  set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; -- 执行Map前进行小文件合并[经过测试,设置上面的参数就能够让小文件进行合并了。]

  输出时合并

  set hive.merge.mapfiles=true; -- 在map-only job后合并文件,默认true

  set hive.merge.mapredfiles=true; -- 在map-reduce job后合并文件,默认false

  set hive.merge.size.per.task=256000000; -- 合并后每个文件的大小,默认256000000

  set hive.merge.smallfiles.avgsize=16000000; -- 平均文件大小,当结果文件小于该值时,将进行合并

数据倾斜

    原因:当我们进行join操作时,会在map端根据key的hash值,shuffle到某一个reduce上去,在reduce端做join连接操作,内存中缓存join左边的表,遍历右边的表,一次做join操作。所以在做join操作时候,将数据量多的表放在join的右边。当数据量比较大,并且key分布不均匀,大量的key都shuffle到一个reduce上了,就出现了数据的倾斜。

    场景

    key为空:当join的key存在空值时,这些值会被hash到同一个reduce中,导致数据倾斜。遇到这样的情况,建议区分空和非空

    count(distinct):数据聚合类的操作sum、count,因为已经在map端做了聚合操作了,到reduce端的数据相对少一些[主要设置参数为:1.set hive.map.aggr=true;默认值,是否在map端进行聚合;2.set hive.groupby.mapaggr.checkinterval = 100000;默认值,在 Map 端进行聚合操作的条目数目],不会出现这个问题。当distinct时,会对group by后的字段和distinct的字段进行hash,如果存在大量重复值,则会导致数据倾斜。

    遇到倾斜时set hive.groupby.skewindata = true;[负载均衡,默认为false。查询计划会有两个 MR Job。第一个 MR Job 中,Map 的输出结果集合会随机分布到 Reduce 中,每个 Reduce 做部分聚合操作,并输出结果,这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce 中,从而达到负载均衡的目的;第二个 MR Job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce 中(这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce 中),最后完成最终的聚合操作]

常用参数设置及说明

--类型:输入时合并小文件

  set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; ----可在map端进行合并,减少小文件

  set mapred.max.split.size=256000000;  ----设置一个合理的数值,最大的切片数据。影响map的个数 

  set mapred.min.split.size.per.node=134217728;  ----一个节点上的最小切片大小。设置为block大小

  set mapred.min.split.size.per.rack=256000000;  ----一个交换机上的最小切片大小

--类型:资源占用量

  set mapreduce.map.memory.mb=5120;

  set mapreduce.reduce.memory.mb=5120;

--类型:设置最大值

  set mapreduce.job.reduces=80;  ----用来对reduer数量进行控制,但设置的过大,不设置效果更好

  set hive.exec.reducers.max=80; ----最大的reduce数

--类型:mapjoin

  set hive.auto.convert.join=true;

  set hive.mapjoin.smalltable.filesize=2500000;

--类型:输出时合并

  set hive.merge.mapfiles=true; -- 在map-only job后合并文件,默认true

  set hive.merge.mapredfiles=true; -- 在map-reduce job后合并文件,默认false

  set hive.merge.size.per.task=256000000; -- 合并后每个文件的大小,默认256000000

  set hive.merge.smallfiles.avgsize=16000000; -- 平均文件大小,当结果文件小于该值时,将进行合并

 --类型:并发及并发数

  set hive.exec.parallel=true;

  set hive.exec.parallel.thread.number=16;

补充

     在写sql的过程中,有时候会用到排序。但是在hive中用order by排序,将带来一个问题。只有一个reduce。往往带来性能的较差。Hive为查询提供了其他排序方式。

  Order by:全局,一个reduce【不建议】

  Distribute by :map的输出排序,排序后,将对应的key通过hash放入到reduce中

  Sort by:局部排序,在每个reduce中实现排序。当reduce数只有一个时,sort by和order by结果没有什么差异【建议】

       Cluster by:分桶中用到,用来指定分桶的字段。cluster by  key = distribute by key sort by key desc  ,是一个简写。cluster by只能降序排序。

 

原创博客,转载请注明出处!欢迎邮件沟通:[email protected]

 

以上是关于hive优化分享的主要内容,如果未能解决你的问题,请参考以下文章

Hive MetaStore 在快手遇到的挑战与优化

hive&hue优化

hive优化方法

Hive优化之Hive的配置参数优化

Hive与优化方法

Hive与优化方法