Hive数据倾斜的示例分析

本文将详细解释Hive数据倾斜的示例分析。边肖觉得挺实用的，分享给大家参考。希望你看完这篇文章能有所收获。

一、离线数据的主要挑战：“数据倾斜”

首先引入了“数据偏斜”的概念。

“倾斜”应该来自统计学中的偏斜分布，数据处理中的倾斜也与此有关。

对于分布式数据处理，我们希望数据均匀分布到各个处理节点，但实际上由于业务数据本身或者分布算法的问题，分布到各个节点的数据量很可能并不是我们预期的那样。

也就是说，只有处理数据最多的节点，才能完成整个数据处理任务，大大降低了时间分布的意义。想想99%的卡住数据。

事实上，即使分配给每个节点的数据量大致相同，数据仍然可能会倾斜。例如，考虑到统计词频的极端问题，如果分配给一个节点的所有单词都是一个单词，则需要很长时间才能显示该节点需要与其他节点具有相同的数据量。

Hive的优化就是利用各种措施和方法来优化处理上述场景的倾斜问题。

00-1010事实上，在Hive SQL的实际开发过程中，实际上只有一小部分Hive SQL的性能与数据倾向有关。

很多时候，Hive SQL运行缓慢是由于开发人员对所使用的数据缺乏了解以及一些不良的使用习惯造成的。

开发人员应确定以下几点：

“需要计算的指标真的需要从数据仓库的公共细节层面自己总结吗？数据公共层团队开发的公共总结层是否已经能够满足自己的需求？通常设计得很好的数据仓库公共层，比如热门、KPI相关的指标，必须已经包含在内，才可以直接使用。

“你真的需要扫描这么多分区吗？比如销售明细交易表，扫描分区一年和扫描分区一周造成的计算和IO成本完全是两个数量级，花费的时间肯定是不一样的。作为开发者，我们需要仔细考虑业务需求，尽量不要浪费计算和存储资源！

"尽量不要使用从表中选择*的方法，并指定使用哪些列。例如，从_表中选择col，col2。此外，在where条件中尽可能添加过滤条件，以删除不相关的数据行，从而减少整个MapReduce任务中要处理和分发的数据量。

"输入文件不应该是大量的小文件。Hive的默认输入拆分是128MB(可配置)，小文件可以先合并成大文件。

保证以上几点后，有时候发现Hive SQL还得长时间运行，甚至不能运行，需要真正的Hive优化技术！

二、Hive 的优化

Hive SQL性能问题基本上大多与join相关，与join无关的问题主要包括group by相关倾斜和count distinct相关优化。

group by 引起的倾斜优化

分组导致的倾斜主要是输入数据列按列分组分布不均匀造成的。

比如假设按照供应商到销售明细的事实单来统计订单数量，那么一些大供应商的订单数量显然非常多，而大部分供应商的订单数量只是平均水平。由于分组依据是根据供应商的标识分配给每个减少任务，因此此时分配给大型供应商的减少任务会分配更多的订单，从而导致数据不对称。

对于由分组引起的倾斜，优化措施非常简单，只需设置以下参数：

set Hive . map . aggr=true set . group by . skew in data=true此时，hive将在数据倾斜时平衡负载，生成的查询计划将有两个MapReduce Job。

在第一个MapReduce作业中，Map的输出结果集将随机分布在Reduce中，每个Reduce都会进行部分聚合操作并输出结果。这种处理的结果是，同一个GroupBy Key可以分配给不同的Reduce，从而达到负载均衡的目的；

第二个MapReduce作业根据GroupBy Key根据预处理后的数据结果分配给Reduce(这个过程可以保证同一个GroupBy Key分配给同一个Reduce)，最后完成最终的聚合操作。

count distinct 优化

在Hive开发过程中，我们应该谨慎使用count distinct，因为这样很容易造成性能问题，比如下面的SQL:

选择计数(区分

t user) from some_table

由于必须去重，因此 Hive 将会把 Map 阶段的输出全部分布到 Reduce Task 上，此时很容易引起性能问题。对于这种情况，可以通过先  group by 再 count 的方式来优化，优化后的 SQL 如下：

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

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

四、大表 Join 小表优化

join 相关的优化主要分为 mapjoin 可以解决的优化 ( 即大表 join 小表) 和 mapjoin 无法解决的优化( 即大表 join 大表  )。大表 join 小表相对容易解决，大表 join 大表相对复杂和难以解决，但也不是不可解决的，只是相对比较麻烦而已。

首先介绍大表 join 小表优化 。仍以销售明细事实表为例来说明大表 join 小表的场景。

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

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

select Seller_srar, count(order_id) as ordre_cnt    from (    select order_id,seller_id         from dwd_sls_fact_detail_table        where partition_value ='20170101'    ) a    Left outer join(    select seller_id,seller_star        from dim_seller        where partition_value='20170101'    ) b    on a.seller_id = b.seller_id    group by b.seller_star;

但正如上述所言，现实世界的二八准则将导致订单集中在部分供应商上，而好的供应商的评级通常会更高，此时更加剧了数据倾斜的程度。如果不加以优化，上述 SQL  将会耗费很长时间，甚至运行不出结果!

通常来说，供应商是有限的，比如上千家、上万家，数据量不会很大，而销售明细事实表比较大，这就是典型的大表 join 小表问题，可以通过 mapjoin  的方式来优化，只需添加 mapjoin hint 即可，优化后的 SQL 如下：

select /*+mapjoin(b)*/ Seller_srar, count(order_id) as ordre_cnt from ( select order_id,seller_id      from dwd_sls_fact_detail_table     where partition_value ='20170101' ) a Left outer join( select seller_id,seller_star     from dim_seller     where partition_value='20170101' ) b on a.seller_id = b.seller_id group by b.seller_star;

/*+mapjoin(b)*/ 即 mapjoin himt，如果需要 mapjoin 多个表，则格式为/*+mapjoin(b,c,d)*/ 。

Hive 对于 mapjoin 是默认开启的，设置参数为：

Set hive.auto.convert.join=ture;

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 )来确定小表的大小是否满足条件(默认 25M)。

• 实际中此参数值所允许的最大值可以修改，但是一般最大不能超过 1GB (太大的话 Map 任务所在的节点内存会撑爆， Hive 会报错 。另外需要注意的是，  HDFS 显示的文件大小是压缩后的大小， 当实际加载到内存的时候，容量会增大很多，很多场景下可能会膨胀 10 倍)。

五、大表 Join 大表优化

如果上述 mapjoin 中小表 dim_seller 很大呢?比如超过了 1GB 的大小?这种就是大表join 大表的问题 。

这类问题相对比较复杂，我们首先引入具体的问题场景，然后基于此介绍各种优化方案。

问题场景

我们先假设一个问题场景：

• A 表为一个汇总表，汇总的是卖家买家最近 N 天交易汇总信息，即对于每个卖家最近 N 天，其每个买家共成交了多少单、总金额是多少，我们这里 N 先只取  90 天，汇总值仅取成交单数 。A 表的字段有：buyer_id 、seller_id 和 pay_cnt_90d 。

• B 表为卖家基本信息表，其中包含卖家的一个分层评级信息，比如把卖家分为 6 个级别：S0、S1、S2、S3、S4、S5、S6 。

要获得的结果是每个买家在各个级别卖家的成交比例信息，比如：

• 某买家 S0:10%; S1:20%; S2:20%; S3:10%; S4:20%; S4:10%; S5:10%。

• B表的字段有：seller_id 和 s_level。

正如 mapjoin 中的例子一样，我们的第一反应是直接 join 表并统计：

select  m.buyer_id  ,sum(pay_cnt_90d) as pay_cnt_90d  ,sum(case when m.s_level=O then pay_cnt_90d end) as pay_cnt_90d_s0  ,sum(case when m.s_level=l then pay_cnt_90d end) as pay_cnt_90d_sl  ,sum(case when m.s_level=2 then pay_cnt_90d end) as pay_cnt_90d_s2  ,sum(case when m.s level=3 then pay cnt 90d end) as pay_cnt_90d_s3 ,sum(case when m.s_level=4 then pay_cnt_90d end) as pay_cnt_90d_s4  ,sum(case when m.s_level=S then pay_cnt_90d end) as pay_cnt_90d_s5  from  ( select  a.buyer_id,a.seller_id,b.s_level,a.pay_cnt_90d  from  ( select buyer_id ,seller_id,pay_cnt_90d  from table A  ) a  join  ( select seller_id,s_level  from table B  ) b  on a.seller_id=b.seller_id  ) m  group by m.buyer_id

但是此 SQL 会引起数据倾斜，原因在于卖家的二八准则。某些卖家 90 天内会有几百万甚至上千万的买家，但是大部分卖家 90 天内的买家数目并不多，  join table_A 和table_B 的时候 ODPS 会按照 Seller_id 进行分发， table_A 的大卖家引起了数据倾斜。

「但是本数据倾斜问题无法用 mapjoin table_B 解决，因为卖家有超过千万条、文件大小几个GB ，超过了 mapjoin 表最大 1GB  的限制。」

方案 1：转化为 mapjoin

大表无法直接mapjoin，那么是否可以间接呢?实际上此思路有两种途径：限制行和限制列。

• 限制行: 不需要join B全表，只需要join其在A表中存在的。对于本问题场景，就是过滤掉 90 天内没有成交的卖家。

• 限制列: 只取需要的字段。

select   m.buyer_id   ,sum(pay_cnt_90d) as pay_cnt_90d   ,sum(case when m.s_level=O then pay_cnt_90d end) as pay_cnt_90d_s0   ,sum(case when m.s_level=l then pay_cnt_90d end) as pay_cnt_90d_sl   ,sum(case when m.s_level=2 then pay_cnt_90d end) as pay_cnt_90d_s2   ,sum(case when m.s level=3 then pay cnt 90d end) as pay_cnt_90d_s3  ,sum(case when m.s_level=4 then pay_cnt_90d end) as pay_cnt_90d_s4   ,sum(case when m.s_level=S then pay_cnt_90d end) as pay_cnt_90d_s5   from   (  select /*+mapjoin(b)*/  a.buyer_id,a.seller_id,b.s_level,a.pay_cnt_90d   from   (  select buyer_id ,seller_id,pay_cnt_90d   from table_A   ) a   join   (  select b0.seller id,s_level   from table_B b0      join      (select seller_id from table_A group by seller_id) a0      on b0.seller_id=a0.seller_id  ) b   on a.seller_id=b.seller_id   ) m   group by m.buyer_id

此方案在一些情况下可以起作用，但很多时候还是无法解决上述问题，因为大部分卖家尽管 90 买家不多 ，但还是有一些的，过滤后的 B 表仍然很大。

方案 2：join 时用 case when 语句

应用场景为: 倾斜的值是明确的而且数量很少，比如null值引起的倾斜。

将这些引起倾斜的值随机分发到Reduce，其主要核心逻辑在于 join 时对这些特殊值concat 随机数,从而达到随机分发的目的。核心逻辑如下：

Select a.user_id,a.order_id,b.user_id    From table_a a    Join table_b b    On (case when a.user_id is null then concat ('hive' ,rand()) else a.user_id end)=b.user_id

Hive已对此进行了优化，不需要修改SQL，只需要设置参数;比如 table_B 的值 "0" 和 "1" 引起倾斜，只需要如下设置：

set hive.optimize.skewinfo=table_B:(seller_id)[("0")("1")];    set hive.optimize.skewjoin=true;

但是方案二还是不能解决上述问题，因为倾斜的卖家大量存在而且动态变化。

方案 3：倍数B表，再取模join

通用方案

是建立一个numbers表,其值只有一列int行，比如从1到10(具体根据倾斜程度确定)，然后放大B表10倍，再取模join。

select    m,buer_id   ,sum(pay_cnt_90d) as pay_cnt_90d    ,sum(case when m.s_level=O then pay_cnt_90d end) as pay cnt 90d so    ,sum(case when m.s_level=l then pay cnt 90d end) as pay cnt 90d_sl    ,sum(case when m.s_level=2 then pay_cnt_90d end) as pay_cnt_90d s2    ,sum(case when m.s_level=3 then pay_cnt_90d end) as pay_cnt_90d_s3    ,sum(case when m.s_level=4 then pay_cnt_90d end) as pay cnt 90d s4    ,sum(case when m.s level=S then pay cnt 90d end) as pay cnt 90d s5    from    (   select    a.buyer_id,a.seller_id,b.s_level,a.pay_cnt_90d    from   (   select buyer_id,seller_id,pay_cnt_90d    from table_A    ) a    JOin    (   select /*+mapjoin(members)*/    seller_id,s_level,member    from table_B    join    numbers    ) b    on a.seller_id=b.seller_id    and mod(a.pay_cnt_90d,10)+1=b.number    ) m    group by m.buyer_id

思路核心在于：既然按照seller_id分发会倾斜，那么再人工增加一列进行分发，这样之前倾斜的值的倾斜程度会减少为原来的1/10。可以通过配置numbers表修改放大倍数来降低倾斜程度，但弊端就是B表会膨胀N倍。

专有方案

通用方案思路是把B表的每条数据都放大了相同的倍数，实际上只需要把大卖家放大倍数即可。

首先需要知道大卖家的名单，即先建立一个临时表动态存放每日最新的大卖家(比如dim_big_seller)，同时此表的大卖家要膨胀预先设定的倍数(比如1000倍)。

在A表和 B表中分别新建一个 join 列，其逻辑为：如果是大卖家，那么 concat 一个随 机分配正整数(0到预定义的倍数之间，本例为0~1000  );如果不是，保持不变。

相比通用方案，专用方案的运行效率明显好了很多，因为只是将B表中大卖家的行数放大了 1000  倍，其他卖家的行数保持不变，但同时也可以看到代码也复杂了很多，而且必须首先建立大卖家表。

方案 4：动态一分为二

实际上方案 2 和 3 都用到了一分为二的思想，但是都不彻底，对于 mapjoin 不能解决的 问题，终极解决方案就是动态一分为  ，即对倾斜的键值和不倾斜的键值分开处理，不倾 斜的正常 join 即可，倾斜的把它们找出来然后做 mapjoin ，最后 union all  其结果即可。

但是此种解决方案比较麻烦，代码会变得复杂而且需要一个临时表存放倾斜的键值。

- 对于 90 天买家数超过 10000 的卖家直接 map join ，对于其他卖家正常 join 即可

- 对于 90 天买家数超过 10000 的卖家直接 map join ，对于其他卖家正常 join 即可 select  m.buyer_id  ,sum(pay_cnt_90d) as pay_cnt_90d  ,sum(case when rn.s_level=O then pay_cnt_90d end) as pay_cnt_90d_s0  ,sum(case when rn.s_level=l then pay_cnt_90d end) as pay_cnt_90d_sl  ,sum(case when rn.s_level=2 then pay_cnt_90d end) as pay_cnt_90d_s2  ,sum(case when rn.s_level=3 then pay_cnt_90d end) as pay_cnt_90d_s3  ,sum(case when rn.s_level=4 then pay_cnt_90d end) as pay_cnt_90d_s4  ,sum(case when rn.s_level=S then pay_cnt_90d end) as pay_cnt_90d_s5 from ( select      a.buyer_id,a.seller_id,b.s_level,a.pay_cnt_90d     from     (     select buyer_id,seller_id,pay_cnt_90d         from table_A     ) a     join     (      select seller_id ,a.s_level         from table_A a         left outer join tmp_table_B b         on a.user_id = b.seller_id         where b.seller_id is null     ) b  on a.seller id=b.seller id  union all  select /*+mapjoin(b)*/  a.buyer_id,a.seller_id,b.s_level,a.pay_cnt_90d  from  select buyer_id,seller_id,pay_cnt_90d  from table A  ) a  join  select seller_id,s_level  from table B  ) b  on a.seller id=b.seller id ) m group by m.buyer_id ) m group by m.byer_id

总结起来，方案 1、2 以及方案 3 中的通用方案不能保证解决大表 join 大表问题，因为它们都存在种种不同的限制和特定的使用场景。

而方案 3 的专用方案和方案 4 是比较推荐的优化方案，但是它们都需要新建一个临时表来存放每日动态变化的大卖家 。

相对方案 4 来说，方案 3 的专用方案不需要对代码框架进行修改，但是 B 表会被放大，所以一定要是维度表，不然统计结果会是错误的 。方案 4  的解决方案最通用，自由度最高，但是对代码的更改也最大，甚至需要更改代码框架，可作为终极方案来使用。

关于“Hive数据倾斜的示例分析”这篇文章就分享到这里了，希望