您的足迹:首页 > Hive >Hive(二)优化总结

Hive(二)优化总结

在介绍hive优化之前,首先要知道hive是什么。

1.hive简介

  • hive是基于hadoop的一哥数据仓库工具,可以将结构化的数据文件映射成一张完整的数据包,并提供完整的sql查询功能,可以将sql语句转化成mapreduce任务进行运行。
  • Hive是建立在 Hadoop 上的数据仓库基础构架。它提供了一系列的工具,可以用来进行数据提取转化加载(ETL),这是一种可以存储、查询和分析存储在 Hadoop 中的大规模数据的机制。

2.Hive与关系型数据库的区别

  • hive和关系数据库存储文件的系统不同,hive使用的是hadoop的HDFS(hadoop的分布式文件系统),关系数据库则是服务器本地的文件系统
  • hive使用的计算模型是mapreduce,而关系数据库则是自己设计的计算模型
  • 关系数据库都是为实时查询的业务进行设计的,而hive则是为海量数据做数据挖掘设计的,实时性很差
  • Hive很容易扩展自己的存储能力和计算能力,这个是继承hadoop的,而关系数据库在这个方面要比数据库差很多

3.Hive优化策略

根据前面介绍,hive是将sql语句转化成MapReduce任务去运行,一个hive查询往往会生成多个Map Reduce Job而一个MR又会有Map,Reduce,Splii,Shuffle,Sort等阶段。所以hive优化可以分为针对MR中具体步骤的优化,MR的整体优化以及查询的优化。

(1)MR中具体步骤的优化 
一个MR Job会有Map、Reduce、Spill、Shuffle和Sort这几个阶段,接下来根据这些具体的阶段去优化hive。

Map的优化: 
Map阶段的优化主要是确定合适的map数。map数的确定与其计算公式有关系,map数的计算公式如下:

num_Map_tasks = max[${Mapred.min.split.size},
                min(${dfs.block.size}, ${Mapred.max.split.size})]
  • 1
  • 2
  • 3

• Mapred.min.split.size指的是数据的最小分割单元大小。 
• Mapred.max.split.size指的是数据的最大分割单元大小。 
• dfs.block.size指的是HDFS设置的数据块大小。

一般来说dfs.block.size这个值是一个已经指定好的值,而且这个参数Hive是识别不到的,所以实际上只有Mapred.min.split.size和Mapred.max.split.size这两个参数(本节内容 
后面就以min和max指代这两个参数)来决定Map数量。在Hive中min的默认值是1B,max的默认值是256MB。所以如果不做修改的话,就是1个Map task处理256MB数据,我们就以调整max为主。通过调整max可以起到调整Map数的作用,减小max可以增加Map数,增大max可以减少Map数。需要提醒的是,直接调整Mapred.Map.tasks这个参数是没有效果的。

调整大小的时机根据查询的不同而不同,总的来讲可以通过观察Map task的完成时间来确定是否需要增加Map资源。如果Map task的完成时间都是接近1分钟,甚至几分钟了,那么往往增加Map数量,使得每个Map task处理的数据量减少,能够让Map task更快完成;而如果Map task的运行时间已经很少了,比如10-20秒,这个时候增加Map不太可能让Map task更快完成,反而可能因为Map需要的初始化时间反而让Job总体速度变慢,这个时候反而需要考虑是否可以把Map的数量减少,这样可以节省更多资源给其他Job。

Reduce的优化: 
Reduce阶段优化的主要工作也是选择合适的Reduce task数量,跟上面的Map优化类似。与Map优化不同的是,Reduce优化时,可以直接设置Mapred.Reduce.tasks参数从而直接指定Reduce的个数。当然直接指定Reduce个数虽然比较方便,但是不利于自动扩展。Reduce数的设置虽然相较Map更灵活,但是也可以像Map一样设定一个自动生成规则,这样运行定时Job的时候就不用担心原来设置的固定Reduce数会由于数据量的变化而不合适。

Hive估算Reduce数量的时候,使用的是下面的公式:

num_Reduce_tasks = min[${Hive.exec.Reducers.max}, 
                      (${input.size} / ${ Hive.exec.Reducers.bytes.per.Reducer})]
  • 1
  • 2
  • 3

也就是说,根据输入的数据量大小来决定Reduce的个数,默认Hive.exec.Reducers.bytes.per.Reducer为1G,而且Reduce个数不能超过一个上限参数值,这个参数的默认取值为999。所以我们可以调整Hive.exec.Reducers.bytes.per.Reducer来设置Reduce个数。 
设置Reduce数同样也是根据运行时间作为参考调整,并且可以根据特定的业务需求、工作负载类型总结出经验,所以不再赘述。

在map阶段和reduce阶段之间会有一个shuffle的过程,一般会有Spill,copy,sort,merge等过程。首先要把Map输出的结果进行排序后做成中间文件,其次这个中间文件就能分发到各个Reduce,最后Reduce端在执行Reduce phase之前把收集到的排序子文件合并成一个排序文件。这个部分可以调的参数挺多,但是一般都是不要调整的,不必重点关注。

Spill与Sort优化: 
在Spill阶段,由于内存不够,数据可能没办法在内存中一次性排序完成,那么就只能把局部排序的文件先保存到磁盘上,这个动作叫Spill,然后Spill出来的多个文件可以在最后进行merge。如果发生Spill,可以通过设置io.Sort.mb来增大Mapper输出buffer的大小,避免Spill的发生。另外合并时可以通过设置io.Sort.factor来使得一次性能够合并更多的数据。调试参数的时候,一个要看Spill的时间成本,一个要看merge的时间成本,还需要注意不要撑爆内存(io.Sort.mb是算在Map的内存里面的)。Reduce端的merge也是一样可以用io.Sort.factor。一般情况下这两个参数很少需要调整,除非很明确知道这个地方是瓶颈。

Copy的优化 
copy阶段是把文件从Map端copy到Reduce端。默认情况下在5%的Map完成的情况下Reduce就开始启动copy,这个有时候是很浪费资源的,因为Reduce一旦启动就被占用,一直等到Map全部完成,收集到所有数据才可以进行后面的动作,所以我们可以等比较多的Map完成之后再启动Reduce流程,这个比例可以通Mapred.Reduce.slowstart.completed.Maps去调整,他的默认值就是5%。如果觉得这么做会减慢Reduce端copy的进度,可以把copy过程的线程增大。tasktracker.http.threads可以决定作为server端的Map用于提供数据传输服务的线程,Mapred.Reduce.parallel.copies可以决定作为client端的Reduce同时从Map端拉取数据的并行度(一次同时从多少个Map拉数据),修改参数的时候这两个注意协调一下,server端能处理client端的请求即可。

(2)文件格式的优化 
文件格式方面有两个问题,一个是给输入和输出选择合适的文件格式,另一个则是小文件问题。 
小文件的问题目前再hive环境下已经得到了比较好的解决,hive的默认配置中就可以在小文件输入时把多个文件合并给一个map处理,输出时如果文件很小也会进行一轮单独的合并。 
文件格式,hive0.9版本有三种,textfile,sequencefile和rcfile。hive1.1后增加了orcfile。总体上来说,orcfile的压缩比例和查询时间会好一旦,推荐使用。

  • textfile: 
    Hive的默认存储格式 
    存储方式:行存储 
    磁盘开销大 数据解析开销大 
    压缩的text文件 hive无法进行合并和拆分
  • sequencefile: 
    二进制文件以key,value的形式序列化到文件中 
    存储方式:行存储 
    可分割 压缩 
    一般选择block压缩 
    优势是文件和Hadoop api中的mapfile是相互兼容的
  • rcfile: 
    存储方式:数据按行分块 每块按照列存储 
    压缩快 快速列存取 
    读记录尽量涉及到的block最少 
    读取需要的列只需要读取每个row group 的头部定义。 
    读取全量数据的操作 性能可能比sequencefile没有明显的优势
  • orcfile: 
    存储方式:数据按行分块 每块按照列存储 
    压缩快 快速列存取 
    效率比rcfile高,是rcfile的改良版本

(3)MR Job的整体优化

Job执行模式: 
Hadoop的Map Reduce Job可以有3种模式执行,即本地模式,伪分布式,还有真正的分布式。本地模式和伪分布式都是在最初学习Hadoop的时候往往被说成是做单机开发的时候用到。但是实际上对于处理数据量非常小的Job,直接启动分布式Job会消耗大量资源,而真正执行计算的时间反而非常少。这个时候就应该使用本地模式执行mr Job,这样执行的时候不会启动分布式Job,执行速度就会快很多。比如一般来说启动分布式Job,无论多小的数据量,执行时间一般不会少于20s,而使用本地mr模式,10秒左右就能出结果。

设置执行模式的主要参数有三个,一个是Hive.exec.mode.local.auto,把他设为true就能够自动开启local mr模式。但是这还不足以启动local mr,输入的文件数量和数据量大小必须要控制,这两个参数分别为Hive.exec.mode.local.auto.tasks.max和Hive.exec.mode.local.auto.inputbytes.max,默认值分别为4和128MB,即默认情况下,Map处理的文件数不超过4个并且总大小小于128MB就启用local mr模式。

Jvm的重用: 
正常情况下,MapReduce启动的JVM在完成一个task之后就退出了,但是如果任务花费时间很短,又要多次启动JVM的情况下(比如对很大数据量进行计数操作),JVM的启动时间就会变成一个比较大的overhead。在这种情况下,可以使用jvm重用的参数: 
set Mapred.Job.reuse.jvm.num.tasks = 5; 
他的作用是让一个jvm运行多次任务之后再退出。这样一来也能节约不少JVM启动时间。

Join算法: 
处理分布式join,一般有两种方法: 
• replication join:把其中一个表复制到所有节点,这样另一个表在每个节点上面的分片就可以跟这个完整的表join了; 
• repartition join:把两份数据按照join key进行hash重分布,让每个节点处理hash值相同的join key数据,也就是做局部的join。

这两种方式在MR Job中分别对应了Map side join和Reduce side join。在一些MPP DB中,数据可以按照某列字段预先进行hash分布,这样在跟这个表以这个字段为join key进行join的时候,该表肯定不需要做数据重分布了,这种功能是以HDFS作为底层文件系统的Hive所没有的。

在默认情况下,Hive的join策略是进行Reduce side join。当两个表中有一个是小表的时候,就可以考虑用Map join了,因为小表复制的代价会好过大表Shuffle的代价。使用Map join的配置方法有两种,一种直接在sql中写hint,语法是/+MapJOIN (tbl)/,其中tbl就是你想要做replication的表。另一种方法是设置Hive.auto.convert.join = true,这样Hive会自动判断当前的join操作是否合适做Map join,主要是找join的两个表中有没有小表。至于多大的表算小表,则是由Hive.smalltable.filesize决定,默认25MB。

但是有的时候,没有一个表足够小到能够放进内存,但是还是想用Map join怎么办?这个时候就要用到bucket Map join。其方法是两个join表在join key上都做hash bucket,并且把你打算复制的那个(相对)小表的bucket数设置为大表的倍数。这样数据就会按照join key做hash bucket。小表依然复制到所有节点,Map join的时候,小表的每一组bucket加载成hashtable,与对应的一个大表bucket做局部join,这样每次只需要加载部分hashtable就可以了。 
然后在两个表的join key都具有唯一性的时候(也就是可做主键),还可以进一步做Sort merge bucket Map join。做法还是两边要做hash bucket,而且每个bucket内部要进行排序。这样一来当两边bucket要做局部join的时候,只需要用类似merge Sort算法中的merge操作一样把两个bucket顺序遍历一遍即可完成,这样甚至都不用把一个bucket完整的加载成hashtable,这对性能的提升会有很大帮助。

首先建表的时候就应该带上bucket:

create table Map_join_test(id int)
clustered by (id) Sorted by (id) into 32 buckets
stored as textfile;
  • 1
  • 2
  • 3
  • 4

然后插入数据,要强制划分为bucket(也就是用Reduce划分hash值相同的数据到相同的文件):

set Hive.enforce.bucketing = true;
insert overwrite table Map_join_test
select * from Map_join_source_data;
  • 1
  • 2
  • 3
  • 4

然后是Map join操作:

普通的map join:
select /*+Mapjoin(a) */count(*)
from Map_join_test a
join Map_join_test b on a.id = b.id;

bucket Map join:
set Hive.Optimize.bucketMapjoin = true;
select count(*)
from Map_join_test a
join Map_join_test b on a.id = b.id;

Sort merge bucket Map joinset Hive.optimize.bucketMapjoin.Sortedmerge = true;
set Hive.input.format = org.apache.Hadoop.Hive.ql.io.BucketizedHiveInputFormat;
select count(*)
from Map_join_test a
join Map_join_test b on a.id = b.id;
本博客所有文章如无特别注明均为原创。作者:数据为王复制或转载请以超链接形式注明转自 数据为王
原文地址《Hive(二)优化总结

相关推荐


  • blogger

发表评论

路人甲 表情
看不清楚?点图切换 Ctrl+Enter快速提交

网友评论(0)