Hive

Hive应用 | Hive性能调优

2021-12-21  本文已影响0人  木衍小哥

Hive作为大数据领域常见的数据仓库组件，在设计和开发阶段需要注意效率。影响Hive效率的不仅仅是数据量过大、数据倾斜、job（小文件过多）或者磁盘I/O过多、MapReduce分配不合理等因素都会对Hive的效率有影响。对Hive的调优可以从架构优化、参数优化以及Hive SQL优化三个方面考虑。

一、架构优化

1、执行引擎

Hive支持多种执行引擎，例如：MR、Tez、Spark等。可以通过hive-site.xml文件中的hive.execution.engine属性配置。

2、优化器

与关系型数据库类型，Hiv在真正执行的时候，会先通过解释器生成AST抽象语法树，然后再通过编译器生成逻辑执行计划，再通过优化器进行优化，优化后通过执行器生成物理执行计划。而Hive有两种优化器：
Vectorize（矢量化优化器）和Cost-Based Optimization（CBO成本优化器）

1）矢量化查询优化（向量化优化器）

矢量化查询执行通过一次批量执行1024行，而不是一行一行来提高扫描、聚合、过滤器和链接等操作的性能，这个功能明显缩短查询执行时间。

-- 默认 false
SET hive.vectorized.execution.enabled = true; 
-- 默认 false
SET hive.vectorized.execution.reduce.enabled = true;

备注：
● 要使用矢量化查询执行，必须用ORC格式存储数据
● 要求执行引擎为Tez

2）成本优化器

Hive的CBO是基于Apache Calcite的，Hive的CBO通过查询成本（有analyze收集的统计信息）会生成有效率的执行计划，最终会较少执行的时间和资源利用，使用CBO的配置如下：

--从 v0.14.0默认为true
SET hive.cbo.enable=true;
-- 默认false 
SET hive.compute.query.using.stats=true; 
-- 默认false
SET hive.stats.fetch.column.stats=true; 
-- 默认true
SET hive.stats.fetch.partition.stats=true;

定期执行表（分析的命令：analyze）的分析，分析后的数据放在元数据库中。
低版本情况下，小表在前的确效率高，高版本优化器已经做了优化。是因为小表的数据可能会放在内存里面，达标的数据内存存不下就会导致效率低。

3、分区表

对于一个比较大的表，将其设计为分区表，可以提升查询的性能，对于一个特定分区的查询，只会加载对应分区路径的数据文件，所以执行速度比较快。

分区字段的选择，避免层级较深的分区，否则会造成太多的子文件夹，常见的分区字段：

4、分桶表

与分区表类似，分桶表的组织方式是将HDFS上的文件分割成多个文件。

分桶可以加快数据采样，也可以提高join的性能，join的字段是分桶字段，因为分桶可以确保某一个key对应的数据在一个特定的桶内（文件），巧妙的选择分桶字段，可以大幅度提升join性能。

通常情况下，分桶字段可以选择经常用过滤操作或者join操作的字段。

5、文件格式

在Hive SQL的创表语句中，可以使用 stored as... 指定表的存储格式。Hive表支持的存储格式有TextFile、SequenceFile、RCFile、ORC、Parquet等。

存储格式一般需要根据业务进行选择，生产环境中绝大多数表都采用TextFile、ORC、Parquet存储格式之一。

6、数据压缩

压缩技术可以减少map与reduce之间的数据传输，从而可以提升查询性能，关于压缩的配置可以在hive命令行中或者hive-site.xml文件中进行配置。

-- 默认式false
SET hive.exec.compress.intermediate=true

开启压缩后，可以选择下面的压缩格式：

image

关于压缩的编码器可以通过mapred-site.xml，hive-site.xml进行配置，也可以通过命令行进行配置，如：

-- 中间结果压缩
SET hive.intermediate.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;
-- 输出结果压缩
SET hive.exec.compress.output=true;
SET mapreduce.output.fileoutputformat.compress.codec = org.apache.hadoop.io.compress.SnappyCodc

二、参数优化

1、本地模式

当Hive处理的数据量较小的时候，启动分布式处理数据就会显得浪费，因为可能启动时间比处理数据时间还要长，Hive支持将作业动态的转为本地模式，需要使用下面的配置：

-- 默认 false
SET hive.exec.mode.local.auto=true; 
-- 默认128M
SET hive.exec.mode.local.auto.inputbytes.max=50000000; 
 -- 默认 4
SET hive.exec.mode.local.auto.input.files.max=5;

一个作业只要满足下面的条件，会启动本地模式：

2、严格模式

所谓严格模式就是不允许执行3种有风险的HQL语句：

  1. 查询分区表的时候不限定分区列的语句

  2. 两个表join产生了笛卡尔积

  3. 用order by 来排序，但是没有指定limit

要开启严格模式，需要将参数要开启严格模式，需要将参数 hive.mapred.mode 设为strict(缺省值)。

该参数可以不在参数文件中定义，在执行SQL之前设置（set hive.mapred.mode=nostrict），即在当前SQL不是严格模式。

3、JVM重用

默认情况下，Hadoop会为一个map或者reduce启动一个JVM，这样可以并行执行map和reduce。当map或者reduce是那种仅运行几秒钟的轻量级作业时，JVM启动进程所耗费的时间会比作业执行的时间还要长。Hadoop可以重用JVM，通过共享JVM以串行而非并行的方式运行map或者reduce。

JVM的重用适用于同一个作业的map和reduce，对于不同作业的task不能够共享JVM。如果要开启JVM重用，需要配置一个作业最大task数量，默认值为1，如果设置为-1，则表示不限制：

-- 代表同一个MR job中顺序执行的5个task重复使用一个JVM，减少启动和关闭的开销 
SET mapreduce.job.jvm.numtasks=5;

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

4、并行执行

Hive的查询通常会被转换成一系列的stage，这些stage之间并不是一直相互依赖的，可以并行执行这些stage，通过下面的方式进行配置：

SET hive.exec.parallel=true; -- 默认false
SET hive.exec.parallel.thread.number=16; -- 默认8

并行执行可以增加集群资源的利用率，如果集群的资源使用率已经很高了，那么并行执行的效果不会很明显。

5、推测执行

在分布式集群环境下，因为程序Bug、负载不均衡、资源分布不均匀等原因，会造成同一个作业的多个任务之间运行速度不一致，有些任务的运行速度可能明显慢于其他任务（比如一个作业的某个任务进度只有50%，而其他所有任务已经运行完成），则这些任务会拖慢作业的整体执行进度。

为了避免这种情况发生，Hadoop采用了推测执行机制，它根据一定的规则推测出“拖后腿”的任务，并为这样的任务启动一个备份任务，让该任务与原始任务同时处理同一份数据，并最终选用最先成功运行完成任务的计算结果作为最终结果。

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

6、合并小文件

-- 缺省参数
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

-- 在 map-only 任务结束时合并小文件，默认true 
SET hive.merge.mapfiles = true;
-- 在 map-reduce 任务结束时合并小文件，默认false 
SET hive.merge.mapredfiles = true;
-- 合并文件的大小，默认256M
SET hive.merge.size.per.task = 268435456;
-- 当输出文件的平均大小小于该值时，启动一个独立的map-reduce任务进行文件merge 
SET hive.merge.smallfiles.avgsize = 16777216;

7、Fetch模式

Fetch模式是指Hive中对某些情况的查询可以不必使用MR计算，select col1, col2 from tab;

可以简单地读取表对应的存储目录下的文件，然后输出查询结果到控制台，在开启Fetch模式之后，在全局查找、字段查找、limit查找等都不启动MR。

-- Default Value: minimal in Hive 0.10.0 through 0.13.1, more in Hive 0.14.0 and later
set hive.fetch.task.conversion=more

三、HiveSQL优化

1、列裁剪和分区裁剪

最基本的操作。所谓列裁剪就是在查询时只读取需要的列，分区裁剪就是读取需要的分区。以用户维度表为例：

SELECT user_id, user_name
FROM dim_user_dd_f
WHERE dt = '2021-10-28' 
and user_level = 1

当列很多或者数据量很大时，如果select *或者不指定分区，全列扫描和全表扫描效率都很低。

Hive中与列裁剪优化相关的配置项是hive.optimize.cp，与分区裁剪优化相关的则是hive.optimize.pruner，默认都是true。在HiveSQL解析阶段对应的则是ColumnPruner逻辑优化器。

2、谓词下推

谓词下推就是将SQL语句中的where谓词逻辑都尽可能提前执行，减少下游处理的数据量。

例如，如下Hive SQL语句：

SELECT a.uid, a.event_type, b.topic_id, b.title
FROM calendar_record_log a
         LEFT OUTER JOIN
     (
         SELECT uid, topic_id, title
         FROM forum_topic
         WHERE pt_date = 20190224
           AND length(content) >= 100
     ) b ON a.uid = b.uid
WHERE a.pt_date = 20190224
  AND status = 0

对forum_topic做过滤的where语句写在子查询内部，而不是外部。Hive中有谓词下推优化的配置项hive.optimize.ppd，默认值true，与它对应的逻辑优化器是PredicatePushDown。该优化器就是将OperatorTree中的FilterOperator向上提，见下图。

image

3、sort by 代替 order by

HiveSQL中的order by 就是将结果按照某个字段进行全局排序，这会导致所有map端数据都进入一个reducer中，在数据量大时可能会长时间计算不完。

如果使用sort by，那么还是会视情况启动多个reducer进行排序，并且保证每个reducer内局部有序。为了控制map端数据分配到reducer的key，往往还要配合distribute by一同使用。如果不加distribute by的话，map端数据就会随机分配到reducer。

举个例子，加入要以UID为key，以上传时间倒序，记录类型倒序输出记录数据：

4、group by 代替 count(distinct)

当要统计某一列的去重数时，如果数据量很大，count(distinct)就会非常慢，原因与order by类似，count(distinct)逻辑只会有很少的reducer来处理。这时可以用group by来改写：

但是这样写会启动两个MR job（单纯distinct只会启动一个），所以要确保数据量大到启动job的overHead远小于计算耗时，才考虑这种方法。当数据集很小或者key的倾斜比较明显时，group by还可能会比distinct慢。

5、group by配置调整---map端预聚合

group by时，如果先起一个combine在map端做部分预聚合，可以有效减少shuffle数据量。

-- 默认为true
set hive.map.aggr = true;

Map端进行聚合操作的条目数

set hive.groupby.mapaggr.checkinterval = 100000;

通过 hive.groupby.mapaggr.checkinterval 参数也可以设置map端预聚合的行 数阈值，超过该值就会分拆job，默认值10W。

上一篇下一篇

猜你喜欢

热点阅读