Hive SQL 参数与性能调优

2022-03-02 00:00:00 数据 执行 文件 分区 分区表


Hive作为大数据平台举足轻重的框架，以其稳定性和简单易用性也成为当前构建企业级数据仓库时使用多的框架之一。

但是如果我们只局限于会使用Hive，而不考虑性能问题，就难搭建出一个完美的数仓，所以Hive性能调优是我们大数据从业者必须掌握的技能。本文将给大家讲解Hive参数与性能调优的一些方法及技巧。

1. Limit 限制调整

一般情况下，limit语句还是需要执行整个查询语句，然后再返回部分结果。

有一个配置属性可以开启，避免这种情况：对数据源进行抽样。

hive.limit.optimize.enable=true -- 开启对数据源进行采样的功能

hive.limit.row.max.size -- 设置小的采样容量

hive.limit.optimize.limit.file -- 设置大的采样样本数

缺点：有可能部分数据永远不会被处理到

2. JOIN优化

1. 使用相同的连接键

当对3个或者更多个表进行join连接时，如果每个on子句都使用相同的连接键的话，那么只会产生一个MapReduce job。

2. 尽量尽早地过滤数据

减少每个阶段的数据量,对于分区表要加分区，同时只选择需要使用到的字段。

3. 尽量原子化操作

尽量避免一个SQL包含复杂逻辑，可以使用中间表来完成复杂的逻辑。

3. 小文件优化

1) 小文件过多产生的影响

  1. 首先对底层存储HDFS来说，HDFS本身就不适合存储大量小文件，小文件过多会导致namenode元数据特别大, 占用太多内存，严重影响HDFS的性能

  2. 对 Hive 来说，在进行查询时，每个小文件都会当成一个块，启动一个Map任务来完成，而一个Map任务启动和初始化的时间远远大于逻辑处理的时间，就会造成很大的资源浪费。而且，同时可执行的Map数量是受限的。

2) 怎么解决小文件过多

1. 使用 hive 自带的 concatenate 命令，自动合并小文件

使用方法：

#对于非分区表
alter table A concatenate;

#对于分区表
alter table B partition(day=20201224) concatenate;

注意：
1、concatenate 命令只支持 RCFILE 和 ORC 文件类型。
2、使用concatenate命令合并小文件时不能指定合并后的文件数量，但可以多次执行该命令。
3、当多次使用concatenate后文件数量不在变化，这个跟参数 mapreduce.input.fileinputformat.split.minsize=256mb 的设置有关，可设定每个文件的小size。

2. 调整参数减少Map数量

设置map输入合并小文件的相关参数：

#执行Map前进行小文件合并
#CombineHiveInputFormat底层是 Hadoop的 CombineFileInputFormat 方法
#此方法是在mapper中将多个文件合成一个split作为输入
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; -- 默认

#每个Map大输入大小(这个值决定了合并后文件的数量)
set mapred.max.split.size=256000000;   -- 256M

#一个节点上split的至少的大小(这个值决定了多个DataNode上的文件是否需要合并)
set mapred.min.split.size.per.node=100000000;  -- 100M

#一个交换机下split的至少的大小(这个值决定了多个交换机上的文件是否需要合并)
set mapred.min.split.size.per.rack=100000000;  -- 100M

设置map输出和reduce输出进行合并的相关参数:

#设置map端输出进行合并，默认为true
set hive.merge.mapfiles = true;

#设置reduce端输出进行合并，默认为false
set hive.merge.mapredfiles = true;

#设置合并文件的大小
set hive.merge.size.per.task = 256*1000*1000;   -- 256M

#当输出文件的平均大小小于该值时，启动一个独立的MapReduce任务进行文件merge
set hive.merge.smallfiles.avgsize=16000000;   -- 16M 

启用压缩:

# hive的查询结果输出是否进行压缩
set hive.exec.compress.output=true;

# MapReduce Job的结果输出是否使用压缩
set mapreduce.output.fileoutputformat.compress=true;

3. 减少Reduce的数量

#reduce 的个数决定了输出的文件的个数，所以可以调整reduce的个数控制hive表的文件数量，
#hive中的分区函数 distribute by 正好是控制MR中partition分区的，
#然后通过设置reduce的数量，结合分区函数让数据均衡的进入每个reduce即可。

#设置reduce的数量有两种方式，种是直接设置reduce个数
set mapreduce.job.reduces=10;

#第二种是设置每个reduce的大小，Hive会根据数据总大小猜测确定一个reduce个数
set hive.exec.reducers.bytes.per.reducer=5120000000; -- 默认是1G，设置为5G

#执行以下语句，将数据均衡的分配到reduce中
set mapreduce.job.reduces=10;
insert overwrite table A partition(dt)
select * from B
distribute by rand();

解释：如设置reduce数量为10，则使用 rand()， 随机生成一个数 x % 10 ，
这样数据就会随机进入 reduce 中，防止出现有的文件过大或过小

4. 使用hadoop的archive将小文件归档

Hadoop Archive简称HAR，是一个高效地将小文件放入HDFS块中的文件存档工具，它能够将多个小文件打包成一个HAR文件，这样在减少namenode内存使用的同时，仍然允许对文件进行透明的访问

#用来控制归档是否可用
set hive.archive.enabled=true;
#通知Hive在创建归档时是否可以设置父目录
set hive.archive.har.parentdir.settable=true;
#控制需要归档文件的大小
set har.partfile.size=1099511627776;

#使用以下命令进行归档
ALTER TABLE A ARCHIVE PARTITION(dt='2022-02-24', hr='12');

#对已归档的分区恢复为原文件
ALTER TABLE A UNARCHIVE PARTITION(dt='2022-02-24', hr='12');

注意:

归档的分区可以查看不能 insert overwrite，必须先 unarchive

4. 本地模式

有时hive的输入数据量是非常小的。在这种情况下，为查询出发执行任务的时间消耗可能会比实际job的执行时间要多的多。对于大多数这种情况，hive可以通过本地模式在单台机器上处理所有的任务。对于小数据集，执行时间会明显被缩短。

set hive.exec.mode.local.auto=true;

当一个job满足如下条件才能真正使用本地模式：

  1. job的输入数据大小必须小于参数：hive.exec.mode.local.auto.inputbytes.max (默认128MB)

  2. job的map数必须小于参数：hive.exec.mode.local.auto.tasks.max (默认4)

  3. job的reduce数必须为0或者1

可用参数 hive.mapred.local.mem (默认0)控制child jvm使用的大内存数。

5. strict模式

开启严格模式对分区表进行查询，在where子句中没有加分区过滤的话，将禁止提交任务(默认：nonstrict)

set hive.mapred.mode=strict 开启严格模式

注：使用严格模式可以禁止以下三种类型的查询：

1. 对分区表的查询必须使用到分区相关的字段

分区表的数据量通常都比较大，对分区表的查询必须使用到分区相关的字段，不允许扫描所有分区，想想也是如果扫描所有分区的话那么对表进行分区还有什么意义呢。

当然某些特殊情况可能还是需要扫描所有分区，这个时候就需要记得确保严格模式被关闭。

2. order by必须带limit

因为要保证全局有序需要将所有的数据拉到一个Reducer上，当数据集比较大时速度会很慢。个人猜测可能是设置了limit N之后就会有一个很简单的优化算法：每个Reducer排序取N然后再合并排序取N即可，可大大减少数据传输量。

3. 禁止笛卡尔积查询（join必须有on连接条件）

Hive不会对where中的连接条件优化为on，所以join必须带有on连接条件，不允许两个表直接相乘。

6. 并行执行优化

Hive会将一个查询转化成一个或者多个阶段。这样的阶段可以是MapReduce阶段、抽样阶段、合并阶段、limit阶段。或者Hive执行过程中可能需要的其他阶段。默认情况下，Hive一次只会执行一个阶段。不过，某个特定的job可能包含众多的阶段，而这些阶段可能并非完全互相依赖的，也就是说有些阶段是可以并行执行的，这样可能使得整个job的执行时间缩短。如果有更多的阶段可以并行执行，那么job可能就越快完成。

通过设置参数hive.exec.parallel值为true，就可以开启并发执行。在共享集群中，需要注意下，如果job中并行阶段增多，那么集群利用率就会增加。

set hive.exec.parallel=true; //打开任务并行执行
set hive.exec.parallel.thread.number=16; //同一个sql允许大并行度，默认为8。

当然得是在系统资源比较空闲的时候才有优势，否则没资源，并行也起不来。

7. JVM优化

JVM重用是Hadoop调优参数的内容，其对Hive的性能具有非常大的影响，特别是对于很难避免小文件的场景或task特别多的场景，这类场景大多数执行时间都很短。

Hadoop的默认配置通常是使用派生JVM来执行map和Reduce任务的。这时JVM的启动过程可能会造成相当大的开销，尤其是执行的job包含有成百上千task任务的情况。JVM重用可以使得JVM实例在同一个job中重新使用N次。N的值可以在Hadoop的mapred-site.xml文件中进行配置。通常在10-20之间，具体多少需要根据具体业务场景测试得出。

<property>
  <name>mapreduce.job.jvm.numtasks</name>
  <value>10</value>
  <description>How many tasks to run per jvm. If set to -1, there is
  no limit. 
  </description>
</property>

我们也可以在Hive中设置：

set mapred.job.reuse.jvm.num.tasks=10 设置jvm重用

这个功能的缺点是，开启JVM重用将一直占用使用到的task插槽，以便进行重用，直到任务完成后才能释放。如果某个“不平衡的”job中有某几个reduce task执行的时间要比其他Reduce task消耗的时间多的多的话，那么保留的插槽就会一直空闲着却无法被其他的job使用，直到所有的task都结束了才会释放。

8. 推测执行优化

在分布式集群环境下，因为程序bug（包括Hadoop本身的bug），负载不均衡或者资源分布不均等原因，会造成同一个作业的多个任务之间运行速度不一致，有些任务的运行速度可能明显慢于其他任务（比如一个作业的某个任务进度只有50%，而其他所有任务已经运行完毕），则这些任务会拖慢作业的整体执行进度。为了避免这种情况发生，Hadoop采用了推测执行（Speculative Execution）机制，它根据一定的法则推测出“拖后腿”的任务，并为这样的任务启动一个备份任务，让该任务与原始任务同时处理同一份数据，并终选用先成功运行完成任务的计算结果作为终结果。

设置开启推测执行参数：Hadoop的mapred-site.xml文件中进行配置：

<property>
  <name>mapreduce.map.speculative</name>
  <value>true</value>
  <description>If true, then multiple instances of some map tasks 
               may be executed in parallel.</description>
</property>

<property>
  <name>mapreduce.reduce.speculative</name>
  <value>true</value>
  <description>If true, then multiple instances of some reduce tasks 
               may be executed in parallel.</description>
</property>

Hive本身也提供了配置项来控制reduce-side的推测执行:

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

关于调优这些推测执行变量，还很难给一个具体的建议。如果用户因为输入数据量很大而需要执行长时间的map或者reduce task的话，那么启动推测执行造成的浪费是非常巨大的。

9. 数据倾斜优化

数据倾斜的原理都知道，就是某一个或几个key占据了整个数据的90%，这样整个任务的效率都会被这个key的处理拖慢，同时也可能会因为相同的key会聚合到一起造成内存溢出。

Hive的数据倾斜一般的处理方案：

常见的做法，通过参数调优：

set hive.map.aggr=true;  
set hive.groupby.skewindata = ture;

当选项设定为true时，生成的查询计划有两个MapReduce任务。

在个MapReduce中，map的输出结果集合会随机分布到reduce中，每个reduce做部分聚合操作，并输出结果。

这样处理的结果是，相同的Group By Key有可能分发到不同的reduce中，从而达到负载均衡的目的；

第二个MapReduce任务再根据预处理的数据结果按照Group By Key分布到reduce中（这个过程可以保证相同的Group By Key分布到同一个reduce中），后完成终的聚合操作。

但是这个处理方案对于我们来说是个黑盒，无法把控。

那么在日常需求的情况下如何处理这种数据倾斜的情况呢：

  1. sample采样，获取哪些集中的key；

  2. 将集中的key按照一定规则添加随机数；

  3. 进行join，由于打散了，所以数据倾斜避免了；

  4. 在处理结果中对之前的添加的随机数进行切分，变成原始的数据。

10. 动态分区调整

动态分区属性：设置为true表示开启动态分区功能（默认为false）

hive.exec.dynamic.partition=true;

动态分区属性：设置为nonstrict,表示允许所有分区都是动态的（默认为strict） 设置为strict，表示必须保证至少有一个分区是静态的

hive.exec.dynamic.partition.mode=strict;

动态分区属性：每个mapper或reducer可以创建的大动态分区个数

hive.exec.max.dynamic.partitions.pernode=100;

动态分区属性：一个动态分区创建语句可以创建的大动态分区个数

hive.exec.max.dynamic.partitions=1000;

动态分区属性：全局可以创建的大文件个数

hive.exec.max.created.files=100000;

11. 其他参数调优

开启CLI提示符前打印出当前所在的数据库名

set hive.cli.print.current.db=true;

让CLI打印出字段名称

hive.cli.print.header=true;

设置任务名称，方便查找监控

set mapred.job.name=P_DWA_D_IA_S_USER_PROD;

决定是否可以在 Map 端进行聚合操作

set hive.map.aggr=true;

有数据倾斜的时候进行负载均衡

set hive.groupby.skewindata=true;

对于简单的不需要聚合的类似SELECT col from table LIMIT n语句，不需要起MapReduce job，直接通过Fetch task获取数据

set hive.fetch.task.conversion=more;

后

代码优化原则：

理透需求原则，这是优化的根本；
把握数据全链路原则，这是优化的脉络；
坚持代码的简洁原则，这让优化更加简单；
没有瓶颈时谈论优化，这是自寻烦恼。

--END--

原文链接：https://mp.weixin.qq.com/s/F2WKvFxQBGlp-6nBuIU6lw

相关文章