hive数据倾斜及处理

火山日常啰嗦

学习了一些大数据的相关框架后，发现应用层的东西确实不难，真正难的都是底层原理，所以我查看了很多资料，借鉴了前人的方法再加上自己的理解，写下了这篇文章。

数据倾斜的直白概念：

数据倾斜就是数据的分布不平衡，某些地方特别多，某些地方又特别少，导致的在处理数据的时候，有些很快就处理完了，而有些又迟迟未能处理完，导致整体任务最终迟迟无法完成，这种现象就是数据倾斜。

针对mapreduce的过程来说就是，有多个reduce，其中有一个或者若干个reduce要处理的数据量特别大，而其他的reduce处理的数据量则比较小，那么这些数据量小的reduce很快就可以完成，而数据量大的则需要很多时间，导致整个任务一直在等它而迟迟无法完成。

跑mr任务时常见的reduce的进度总是卡在99%，这种现象很大可能就是数据倾斜造成的。

产生数据倾斜的原因：

1） key的分布不均匀或者说某些key太集中。

上面就说过，reduce的数据量大小差异过大，而reduce的数据是分区的结果，分区是对key求hash值，根据hash值决定该key被分到某个分区，进而进入到某个reduce，而如果key很集中或者相同，那么计算得到它们的hash值可能一样，那么就会被分配到同一个reduce，就会造成这个reduce所要处理的数据量过大。

2） 业务数据自身的特性。

比如某些业务数据作为key的字段本就很集中，那么结果肯定会导致数据倾斜啊。

还有其他的一些原因，但是，根本原因还是key的分布不均匀，而其他的原因就是会造成key不均匀，进而导致数据倾斜的后果，所以说根本原因是key的分布不均匀。

既然有数据倾斜这种现象，就必须要有数据倾斜对应的处理方案啊。

简单地说数据倾斜这种现象导致的任务迟迟不能完成，耗费了太多时间，极大地影响了性能，所以我们数据倾斜的解决方案设计思路就是往如何提高性能，即如何缩短任务的处理时间这方面考虑的，而要提高性能，就要让key分布相对均衡，所以我们的终极目标就是考虑如何预处理数据才能够使得它的key分布均匀。

常见的数据倾斜处理方案：

1 设置参数

1）设置hivemapaggr=true //开启map端部分聚合功能，就是将key相同的归到一起，减少数据量，这样就可以相对地减少进入reduce的数据量，在一定程度上可以提高性能，当然，如果数据的减少量微乎其微，那对性能的影响几乎没啥变化。

2）设置hivegroupbyskewindata=true //如果发生了数据倾斜就可以通过它来进行负载均衡。当选项设定为 true，生成的查询计划会有两个 MR Job。第一个 MR Job 中，Map 的输出结果集合会随机分布到 Reduce 中，每个 Reduce 做部分聚合 *** 作，并输出结果，这样处理的结果是相同的Key 有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；第二个 MR Job 再根据预处理的数据结果按照Key 分布到 Reduce 中（这个过程是按照key的hash值进行分区的，不同于mr job1的随机分配，这次可以保证相同的Key 被分布到同一个 Reduce 中），最后完成最终的聚合 *** 作。所以它主要就是先通过第一个mr job将key随机分配到reduce，使得会造成数据倾斜的key可能被分配到不同的reduce上，从而达到负载均衡的目的。到第二个mr job中，因为第一个mr job已经在reduce中对这些数据进行了部分聚合（就像单词统计的例子，a这个字母在不同的reduce中，已经算出它在每个reduce中的个数，但是最终的总的个数还没算出来，那么就将它传到第二个mr job，这样就可以得到总的单词个数），所以这里直接进行最后的聚合就可以了。

3）hiveexecreducersbytesperreducer=1000000000 （单位是字节）

每个reduce能够处理的数据量大小，默认是1G

4）hiveexecreducersmax=999

最大可以开启的reduce个数，默认是999个

在只配了hiveexecreducersbytesperreducer以及hiveexecreducersmax的情况下，实际的reduce个数会根据实际的数据总量/每个reduce处理的数据量来决定。

5）mapredreducetasks=-1

实际运行的reduce个数，默认是-1，可以认为指定，但是如果认为在此指定了，那么就不会通过实际的总数据量/hiveexecreducersbytesperreducer来决定reduce的个数了。

2 sql语句优化

给几个具体的场景以及在这些场景下的处理方案：

1）进行表的join这种业务 *** 作时，经常会产生数据倾斜。

原因就是这些业务数据本就存在key会分布不均匀的风险，所以我们join时不能使用普通的join（reduce端join）或者可以使用普通join，但是是优化后的。

但是这种 *** 作有个前提条件就是仅适用于小表join大表，而小表怎么定义它的大小，多小的表才算小表，这里有个参数可以确定的（但是这个参数名我暂时忘记了），如果小表的数据大小小于这个值，就可以使用map join，而是在这种情况下是自动使用map join这种方案的。所以如果是大小表join，直接用map join，避免数据倾斜。

方法1：（普通join）

select from log a join users b on （auser_id is not null and auser_id = buser_id ）；

这是属于表的内连接的，两张表不满足条件的记录都不保留。

方法2：检测到user_id是null时给它赋予一个新值（这个新值由一个字符串（比如我自己给它定一个 hive）加上一个随机数组成），这样就可以将原来集中的key分散开来，也避免了数据倾斜的风险。而且因为这些数据本来就是无效数据，根本不会出现在结果表中，所以，这样处理user_id（由一个字符串（比如我自己给它定一个 hive）加上一个随机数），它也无法关联的，因为有效的数据的user_id没有这种形式的，所以就算这些无效数据出现在不同的reduce中还是不会影响结果的，我这样处理只是为了将它们分散开而已，所以用这种方法处理，结果表中也不会出现null这些无效数据，跟过滤处理方案得到的结果是一样的。（普通join）

select

from log a

join users b

on case when auser_id is null then concat(‘hive’,rand() ) else auser_id end = buser_id;

但是这两种方案只是适用于大表join大表的内连接，两张表的无效数据都不保留。

但是如果对于左外连接或者右外连接这种情况，即使驱动表中某些记录在另一张表中没有数据与它对应，但我们是依然需要保留驱动表的这些数据的，那该怎么办呢？其实很简单，只需要将上述方法得到的结果再与驱动表的这些无数据取并集就可以了。

如下：

select from log a

left outer join users b

on auser_id is not null

and auser_id = buser_id

union all

select from log a

where auser_id is null;

2)虽然都是大表，但是呢对于某些业务数据而言，其有用的部分只占它所在表的很少一部分，那么我们就可以将它们先取出来，得到的结果应该是一张小表，那么就可以使用map join来避免数据倾斜了。

场景：用户表中user_id字段为int，log表中user_id字段既有string类型也有int类型。

当按照user_id进行两个表的Join *** 作时，因为我们在连接时要进行user_id的比较，所以需要user_id的类型都相同，如果我们选择将log表中的String类型转换为int类型，那么就可能会出现这种情况：String类型转换为int类型得到的都是null值（这就是类型转换的问题了，String类型数据转换为int类型会失败，数据丢失，就会赋null值），如果所有的String类型的user_id都变成了null,那么就又出现了集中的key，分区后就又会导致数据倾斜。所以我们进行类型转换时不能选择将String类型转换为int，而应该将int类型转换为String,因为int转换为String不会出问题，int类型原来的值是什么，转换为String后对应的字符串就会是什么，形式没变，只是类型变了而已。

解决方法：把int类型转换成字符串类型

select from users a

join logs b

on (ausr_id = cast(buser_id as string));

比如有一份日志，要你从日志中统计某天有多少个用户访问网站，即统计有多少个不同的user_id；但是呢这个网站却又恰巧遭到攻击，日志中大部分都是同一个user_id的记录，其他的user_id属于正常访问，访问量不会很大，在这种情况下，当你直接使用count(distinct user_id)时，这也是要跑mr任务的啊，这时这些大量的相同的user_id就是集中的key了，结果就是通过分区它们都被分到一个reduce中，就会造成这个reduce处理的数据特别大，而其中的reduce处理的数据都很小，所以就会造成数据倾斜。

那么要怎么优化呢？

方法1：可以先找出这个user_id是什么，过滤掉它，然后通过count(distinct user_id)计算出剩余的那些user_id的个数，最后再加1（这1个就是那个被过滤掉的user_id，虽然它有大量的记录，但是ser_id相同的都是同一个用户，而我们要计算的就是用户数）

sql语句展示：

分组求和后降序排序，就可以得到这个数据量最大的user_id是什么，然后我们下一步 *** 作时就过滤它，等计算完其他的再加上它这一个。

select user_id,count(user_id) from log group by user_id desc limit 2;

select count（distinct user_id）+1 as sum from log;

sum就是最终的结果--用户数

方法2：我们可以先通过group by分组，然后再在分组得到的结果的基础之上进行count

sql语句展示：

select count() from (select user_id from log group by user_id) new_log;

总的来说就是，数据倾斜的根源是key分布不均匀，所以应对方案要么是从源头解决（不让数据分区，直接在map端搞定），要么就是在分区时将这些集中却无效的key过滤（清洗）掉，或者是想办法将这些key打乱，让它们进入到不同的reduce中。

性能调优是指通过调整使得机器处理任务的速度更快，所花的时间更少，而数据倾斜的处理是hive性能调优的一部分，通过处理能够大大减少任务的运行时间。

除了数据倾斜的处理之外，hive的优化还有其他方面的，例如where子句优化：

select from a left outer join b on (akey=bkey) where adate='2017-07-11' and bdate='2017-07-11';

这是一个左外连接。

这个sql语句执行的结果是：得到的结果是表a与表b的连接表，且表中的记录的date都是'2017-07-11'。

而这个sql语句的执行过程是：逐条获取到a表的记录，然后扫描b表，寻找字段key值为akey的记录,找到后将b表的这条记录连接到a表上，然后判断连接后的这条记录是否满足条件adate='2017-07-11' and bdate='2017-07-11'，如果满足，则显示，否则，丢弃。

因为这是一个左外连接，且a为驱动表，连接时在a中发现key而在b中没有发现与之相等的key时，b中的列将置为null，包括列date，一个不为null，一个为null，这样后边的where就没有用了。

简答的说这个方案的做法就是先按连接条件进行连接，连接后再看where条件，如果不满足就丢弃，那之前连接所做的那些功夫就浪费了，白白耗费了资源（cpu等），增加了运行的总时间，如果有一种方案可以在未进行连接之前就直接判断出不满足最终的条件，那么就可以直接丢弃它，这样对于这样的记录就不要浪费资源以及时间去连接了，这样也是能提升性能的，下面就看看这种方案：

sql语句：

将刚才的where限制条件直接放到on里面，那么就变成了满足这三个条件才会进行连接，不满足的直接过滤掉，就像上面所说的，少了无效连接那一步，就相对地节约了时间，如果这样的无效连接的记录很多的话，那么采用这种改进版的方案无疑能够较大程度地提高性能。

select from a left outer join b on (akey=bkey and adate='2017-07-11' and bdate='2017-07-11'）;

不管怎么说，我们在运行任务时，总是希望能加快运行速度，缩短运行时间，更快地得到结果，即提升性能，这是我们的目的，这就是我们所谓的性能调优。

关于小表join大表的补充：

表join时的 *** 作是这样的：

当 *** 作到驱动表的某条记录时，就会全局扫描另一张表，寻找满足条件的记录，而当扫描它时，为了读取速度更快，一般都选先将它加载到内存，而内存的大小是有限的，为了不占据过多的内存或者避免内存溢出，加载进入内存的表一般是小表，即数据量比较小，map join就是这样做的。

即驱动表不放进内存，而另一张表（即要连接到驱动表的那一张表）就要先加载进内存，为了扫描速度更快，提高性能。

比如select from a left outer join b on (akey=bkey);

左外连接，驱动表是a，表b的记录是要被连接到表a的，所以每在a上连接一条记录就要被全局扫描一次的表是b，所以表b应先加载到内存（前提是表b是小表，如果是大表的话，估计会产生oom异常--out of memory内存溢出异常）。

select from aa right outer join bb on (akey=bkey);

右外连接，驱动表是bb，aa应先加载到内存（前提是小表）。

ps：希望我的分享能帮助到有需要的伙伴哦。我不是大神的哦，如果文中有误，还请大家不吝赐教，帮忙指正，谢谢了！！！

使用Datedif(日期1，日期2，"m")函数DATEDIF是EXCEL中的函数，还有以下的使用方法一并教教：1、简要说明：返回两个日期之间的年\月\日间隔数2、基本语法：=DATEDIF(开始日期，结束日期，单位代码）3、实例1：题目：计算出生日期为1973-4-1人的年龄公式：=DATEDIF("1973-4-1",TODAY(),"Y")结果：33简要说明当单位代码为"Y"时,计算结果是两个日期间隔的年数4、实例2：题目：计算日期为1973-4-1和当前日期的间隔月份数公式：=DATEDIF("1973-4-1",TODAY(),"M")结果：403简要说明当单位代码为"M"时,计算结果是两个日期间隔的月份数5、实例3：题目：计算日期为1973-4-1和当前日期的间隔天数公式：=DATEDIF("1973-4-1",TODAY(),"D")结果：12273简要说明当单位代码为"D"时,计算结果是两个日期间隔的天数5、实例4：题目：计算日期为1973-4-1和当前日期的不计年数的间隔天数公式：=DATEDIF("1973-4-1",TODAY(),"YD")结果：220简要说明当单位代码为"YD"时,计算结果是两个日期间隔的天数忽略年数差5、实例5：题目：计算日期为1973-4-1和当前日期的不计月份和年份的间隔天数公式：=DATEDIF("1973-4-1",TODAY(),"MD")结果：6简要说明当单位代码为"MD"时,计算结果是两个日期间隔的天数忽略年数和月份之差5、实例6：题目：计算日期为1973-4-1和当前日期的不计年份的间隔月份数公式：=DATEDIF("1973-4-1",TODAY(),"YM")结果：7简要说明当单位代码为"YM"时,计算结果是两个日期间隔的月份数不计相差年数

大家好呀，这节课学习 HiveSQL 的常用优化技巧。由于 Hive 主要用来处理非常大的数据，运行过程由于通常要经过 MapReduce 的过程，因此不像 MySQL 一样很快出结果。而使用不同方法写出来的 HiveSQL 语句执行效率也是不一样的，因此为了减少等待的时间，提高服务器的运行效率，我们需要在 HiveSQL 的语句上进行一些优化。

本节课的主要内容 ：

引言

1、技巧一：列裁剪和分区裁剪

（1）列裁剪

（2）分区裁剪

2、技巧二：排序技巧——sort by代替order by

3、技巧三：去重技巧——用group by来替换distinct

4、技巧四：聚合技巧——grouping sets、cube、rollup

（1）grouping sets

（2）cube

（3）rollup

5、技巧五：换个思路解题

6、技巧六：union all时可以开启并发执行

7、技巧七：表连接优化

8、技巧八：遵循严格模式

Hive 作为大数据领域常用的数据仓库组件，在平时设计和查询时要特别注意效率。影响Hive效率的几乎从不是数据量过大，而是数据倾斜、数据冗余、job 或 I/O 过多、MapReduce 分配不合理等等。对 Hive 的调优既包含对HiveSQL 语句本身的优化，也包含 Hive 配置项和 MR 方面的调整。

列裁剪就是在查询时只读取需要的列。当列很多或者数据量很大时，如果select 所有的列或者不指定分区，导致的全表扫描和全分区扫描效率都很低。Hive中与列裁剪优化相关的配置项是 hiveoptimizecp ，默认是 true 。

分区裁剪就是在查询时只读需要的分区。Hive中与分区裁剪优化相关的则是 hiveoptimizepruner ，默认是 true 。

HiveSQL中的 order by 与其他 SQL 语言中的功能一样，就是将结果按某个字段全局排序，这会导致所有map端数据都进入一个 reduce 中，在数据量大时可能会长时间计算不完。

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

这里需要解释一下， distribute by 和 sort by 结合使用是如何相较于 order by 提升运行效率的。

假如我们要对一张很大的用户信息表按照年龄进行分组，优化前的写法是直接 order by age 。使用 distribute by 和 sort by 结合进行优化的时候， sort by 后面还是 age 这个排序字段， distribute by 后面选择一个没有重复值的均匀字段，比如 user_id 。

这样做的原因是，通常用户的年龄分布是不均匀的，比如20岁以下和50岁以上的人非常少，中间几个年龄段的人又非常多，在 Map 阶段就会造成有些任务很大，有些任务很小。那通过 distribute by 一个均匀字段，就可以让系统均匀地进行“分桶”，对每个桶进行排序，最后再组合，这样就能从整体上提升 MapReduce 的效率。

取出 user_trade 表中全部支付用户：

原有写法的执行时长：

优化写法的执行时长：

考虑对之前的案例进行优化：

注意： 在极大的数据量（且很多重复值）时，可以先 group by 去重，再 count() 计数，效率高于直接 count(distinct ) 。

如果我们想知道用户的性别分布、城市分布、等级分布，你会怎么写？

通常写法：

缺点 ：要分别写三次SQL，需要执行三次，重复工作，且费时。

那该怎么优化呢？

注意 ：这个聚合结果相当于纵向地堆在一起了（Union all），分类字段用不同列来进行区分，也就是每一行数据都包含 4 列，前三列是分类字段，最后一列是聚合计算的结果。

GROUPING SETS() ：在 group by 查询中，根据不同的维度组合进行聚合，等价于将不同维度的 group by 结果集进行 union all。聚合规则在括号中进行指定。

如果我们想知道用户的性别分布以及每个性别的城市分布，你会怎么写？

那该怎么优化呢？

注意： 第二列为NULL的，就是性别的用户分布，其余有城市的均为每个性别的城市分布。

cube：根据 group by 维度的所有组合进行聚合

注意 ：跑完数据后，整理很关键！！！

rollup：以最左侧的维度为主，进行层级聚合，是cube的子集。

如果我想同时计算出，每个月的支付金额，以及每年的总支付金额，该怎么办？

那应该如何优化呢？

条条大路通罗马，写SQL亦是如此，能达到同样效果的SQL有很多种，要学会思路转换，灵活应用。

来看一个我们之前做过的案例：

有没有别的写法呢？

Hive 中互相没有依赖关系的 job 间是可以并行执行的，最典型的就是

多个子查询union all。在集群资源相对充足的情况下，可以开启并

行执行。参数设置： set hiveexecparallel=true;

时间对比：

所谓严格模式，就是强制不允许用户执行3种有风险的 HiveSQL 语句，一旦执行会直接报错。

要开启严格模式，需要将参数 hivemapredmode 设为 strict 。

好啦，这节课的内容就是这些。以上优化技巧需要大家在平时的练习和使用中有意识地去注意自己的语句，不断改进，就能掌握最优的写法。

目标：

1、掌握HQL中的各种连接及其组合使用；

2、掌握数据分析中业务指标思路转换的技巧；

3、区分好full join 和 union all 的使用场景；

4、在多表连接时，注意各种细节和业务逻辑；

5、复杂表连接要学会分步骤处理

需注意：

1、表连接时，必须进行重命名；

2、on后面使用的连接条件必须起到 唯一键值 的作用（有时会有多个字段组合）；

3、inner可省略不写，效果是一样的

4、表连接时不能使用 a join b join c这种方式，不然会极度浪费电脑的资源和延长查询时间，要在子查询的表里先做好筛选之后在连接；

1）找出在2019年购买后又退款的用户（记得要去重）

注意：一定要先去重，再做表连接，养成良好的习惯（虽然可以先连接再去重，但是那么做会使执行效率很低）

2）在2017年和2018年都购买的用户

3）在2017年、2018年、2019年都有交易的用户

进行左连接之后，以左表为全集，返回能够匹配上的右边表的匹配结果，没有匹配上的则显示NULL。

拓展：

right join：以右表为全集，返回能够匹配上的左边表的匹配结果，没有匹配上的则显示NULL，可以由left join改写出同样的结果。

4）在2019年购买，但是没有退款的用户

5）在2019年由购买的用户的学历分布

6）在2017年和2018年都购买，但是没有在2019年购买的用户

查询两个表的所有用户时使用full join是一个比较好的方法（需要用到coalesce函数：

注：coalesce函数，coalesce(expression1,expression2,,expression n)，依次参考各参数表达式，遇到非null值即停止并返回该值，如果所有的表达式都是空值，最终将返回一个空值。

注：表合并时字段名称必须一致，字段顺序必须一致，而且不用填写连接条件

7）2017-2019年由交易的所有用户数

union all 和 union 的区别：

union all 不会去重，不会排序，效率较快；union 会去重且排序，效率较慢。

如果表很大时，推荐先去重，再进行 union all ，不能直接对表进行 union all，不然效率很慢。

8）2019年每个用户的支付和退款金额汇总

也可以使用 full join 的方式：

9）2019年每个支付用户的支付金额和退款金额

10）首次激活时间在2017年，但是一直没有支付的用户年龄段分布

步骤总结：

1、先筛选出年份为2017注册的用户；

2、没有支付的人；

3、年龄段分布

注意：由于age也是在user_info的表格里，第三步用的字段需要在第一步进行预处理，所以在限制时间的时候需要同时对年龄段进行预处理，这样在第三步的时候才会由年龄段这个字段；需要注意对 case when 的字段进行重命名才能进行后续的 *** 作

11）2018、2019年交易的用户，其激活时间段分布

步骤总结：

1 取出2018和2019年所有的交易用户的交集

2 取出所有用户的激活时间

3 统计时间分布

select orderid,fenjian,timee

from

(

select orderid,fenjian,timee,row_number(orderid,fenjian) rn

from (

select orderid,fenjian,timee from tableName

distribute by orderid,fenjian sort by orderid,fenjian,timee asc

) t1

) t2

where t2rn=1

以上就是关于hive数据倾斜及处理全部的内容，包括:hive数据倾斜及处理、Hivesql计算两个时间戳相差的分钟数、数据分析课程笔记 - 19 - HiveSQL 常用优化技巧等相关内容解答，如果想了解更多相关内容，可以关注我们，你们的支持是我们更新的动力！