Hive解决数据倾斜的各种优化方法

超、凢脫俗 2023-01-08 08:27 167阅读 0赞

# 一、概念 #

**数据处理中的数据倾斜**：个人理解，在数据处理的MapReduce程序中，由于数据的特殊性，数据中存在大量相同key的数据，根据业务需求需要对这个key进行分区操作（group by/join）时，在map的partition阶段将大数据量的相同key的数据全部分配到同一个Reduce，导致Reduce的节点数据量分配极度不均衡的现象，称为数据倾斜。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzE2MTgxMQ_size_16_color_FFFFFF_t_70][]

**数据倾斜有哪些表现：**

*  最直观的表现就是：Hive SQL运行得慢
 *  任务进度长时间维持在99%（或100%），查看任务监控页面，发现只有少量（1个或几个）reduce子任务未完成，因为其处理的数据量和其他reduce差异过大。
 *  单一reduce的记录数与平均记录数差异过大，通常可能达到3倍甚至更多，最长时长远大于平均时长。

**数据倾斜的原因：**

1.  key分布不均匀
2.  业务数据本身的特性
3.  建表时考虑不周
4.  某些SQL语句本身就有数据倾斜

# 二、数据倾斜优化方法分类 #

在实际Hive SQL开发的过程中，Hive SQL性能的问题上实际只有一小部分和数据倾斜相关。很多时候，Hive SQL运行得慢是由开发人员对于使用的数据了解不够以及一些不良的使用习惯引起的，我们可以确定一些关键点，看看是否能通过业务层面来避免写这中运行的特别慢的 hive sql，比如使用公共汇总层的数据代替公共明细层的数据。除此之外就需要真正的Hive优化技术了。

**所以个人将优化方法分为以下三大类：**

1.  业务优化
2.  join无关的优化
    
    1.  group by 引起的数据倾斜优化
    2.  count distinct 优化
3.  join相关的优化
    
    1.  mapjoin可以解决的join优化(即大表join小表)
    2.  mapjoin无法解决的join优化(即大表join大表)

# 三、具体的优化方法 #

## 1.业务优化 ##

很多时候，Hive SQL运行得慢是由开发人员对于使用的数据了解不够以及一些不良的使用习惯引起的。

开发人员需要确定以下几点。

*  需要计算的指标真的需要从数据仓库的公共明细层来自行汇总么？是不是数据公共层团队开发的公共汇总层已经可以满足自己的需求？对于大众的、KPI相关的指标等通常设计良好的数据仓库公共层肯定已经包含了，直接使用即可。
 *  真的需要扫描这么多分区么？比如对于销售明细事务表来说，扫描一年的分区和扫描一周的分区所带来的计算、IO开销完全是两个量级，所耗费的时间肯定也是不同的。笔者并不是说不能扫描一年的分区，而是希望开发人员需要仔细考虑业务需求，尽量不浪费计算和存储资源，毕竟大数据也不是毫无代价的。
 *  尽量不要使用select \* from your\_table这样的方式，用到哪些列就指定哪些列，如select col1, col2 from your\_table。另外，where条件中也尽量添加过滤条件，以去掉无关的数据行，从而减少整个MapReduce任务中需要处理、分发的数据量。
 *  输入文件不要是大量的小文件。Hive的默认Input Split是128MB（可配置），小文件可先合并成大文件。

在保证了上述几点之后，有的时候发现Hive SQL还是要运行很长时间，甚至运行不出来，这时就需要真正的Hive优化技术了。

## 2.join无关的优化 ##

Hive SQL性能问题基本上大部分都和join相关，对于和join无关的问题主要有group by相关的倾斜和count distinct相关的优化。

### 1）group by 引起的数据倾斜优化 ###

group by引起的倾斜主要是输入数据行按照group by列分布不均匀引起的。

比如，假设按照供应商对销售明细事实表来统计订单数，那么部分大供应商的订单量显然非常多，而多数供应商的订单量就一般，由于group by的时候是按照供应商的ID分发到每个ReduceTask，那么此时分配到大供应商的Reduce Task就分配了更多的订单，从而导致数据倾斜。

对于group by引起的倾斜，优化措施非常简单，只需设置下面参数即可：

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

此时Hive在数据倾斜的时候会进行负载均衡，生成的查询计划会有两个MapReduce Job。

第一个MapReduce Job中，Map的输出结果集合会随机分布到Reduce中，每个Reduce做部分聚合操作并输出结果，这样处理的结果是相同的GroupBy Key有可能被分布到不同的Reduce中，从而达到负载均衡的目的；

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

### 2）count distinct优化 ###

在Hive开发过程中，应该小心使用count distinct，因为很容易引起性能问题，比如下面的SQL：

select count(distinct user) from some_table;

由于必须去重，因此Hive将会把Map阶段的输出全部分布到一个Reduce Task上，此时很容易引起性能问题。

对于这种情况，可以通过先group by再count的方式来优化，优化后的SQL如下

select count(*) 
    from
    (select user from some_table group by user) a;

其原理为：利用group by去重，再统计group by的行数目。

## 3.join相关的优化 ##

### 1）mapjoin可以解决的join优化(即大表join小表) ###

**背景**：通常情况下，JOIN操作在Reduce阶段执行表连接。整个JOIN过程包含Map、Shuffle、Reduce三个阶段。MAPJOIN在Map阶段执行表连接，而非等到Reduce阶段才执行表连接。这样就节省了大量数据传输的时间以及系统资源，从而起到了优化作业的作用。在大表和一个或多个小表JOIN的场景下，MAPJOIN会将您指定的小表全部加载到执行JOIN操作的程序的内存中，因此指定的表仅能为小表。

以销售明细事实表为例来说明大表join小表的场景。

假如供应商会进行评级，比如（五星、四星、三星、两星、一星），此时业务人员希望能够分析各供应商星级的每天销售情况及其占比。

开发人员一般会写出如下SQL：

select
        seller_star,
        count(order_id) as order_cnt
    from 
    (select order_id,seller_id from detail_table where dt=20210119) a
    left join 
    (select seller_id,seller_star from dim_seller where dt=20210119) b
    on a.seller_id = b.seller_id
    group by b.seller_star

但正如上述所言，现实世界的二八准则将导致订单集中在部分供应商上，而好的供应商的评级通常会更高，此时更加剧了数据倾斜的程度，如果不加以优化，上述SQL将会耗费很长时间，甚至运行不出结果。通常来说，供应商是有限的，比如上千家、上万家，数据量不会很大，而销售明细事实表比较大，这就是典型的大表join小表问题，可以通过mapjoin的方式来优化，只需添加mapjoin hint即可，优化后的SQL如下：

select  /*+mapjoin(b)*/ 
        seller_star,
        count(order_id) as order_cnt
    from 
    (select order_id,seller_id from detail_table where dt=20210119) a
    left join 
    (select seller_id,seller_star from dim_seller where dt=20210119) b
    on a.seller_id = b.seller_id
    group by b.seller_star

/\*+mapjoin(b)\*/即mapjoin hint，如果需要mapjoin多个表，则格式为/\*+mapjoin(b, c, d)\*/。Hive对于mapjoin是默认开启的，设置参数为：

set hive.auto.convert.join=true;

mapjoin优化是在Map阶段进行join，而不是像通常那样在Reduce阶段按照join列进行分发后在每个Reduce任务节点上进行join，不需要分发也就没有倾斜的问题，相反Hive会将小表全量复制到每个Map任务节点（对于本例是dim\_seller表，当然仅全量复制b表sql指定的列），然后每个Map任务节点执行lookup小表即可。

所以，小表不能太大，否则全量复制分发得不偿失，实际上Hive根据参数hive.mapjoin.smalltable.filesize（0.11.0版本后是hive.auto.convert.join.noconditionaltask.size）来确定小表的大小是否满足条件（默认25MB），实际中此参数值所允许的最大值可以修改，但是一般最大不能超过1GB（太大的话Map任务所在的节点内存会撑爆，Hive会报错。另外需要注意的是，HDFS显示的文件大小是压缩后的大小，当实际加载到内存的时候，容量会增大很多，很多场景下可能会膨胀10倍）。

**使用注意事项：**

*  使用MAPJOIN时，在引用小表或子查询时，需要引用别名。
 *  MAPJOIN支持小表为子查询。
 *  LEFT OUTER JOIN的左表必须是大表。
 *  RIGHT OUTER JOIN的右表必须是大表。
 *  INNER JOIN的左表或右表均可以作为大表。
 *  目前，MaxCompute在MAPJOIN中最多支持指定128张小表，否则报语法错误。MAPJOIN中多个小表用逗号隔开，例如`/*+MAPJOIN(a,b,c)*/`。

### 2）mapjoin无法解决的join优化(即大表join大表) ###

//TODO

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzE2MTgxMQ_size_16_color_FFFFFF_t_70]: /images/20221119/2b117ce4881c41ba9e71a6de8252f06e.png