大表与大表join数据倾斜_spark面试必问|碰到数据倾斜你该咋办

目录&＃xff1a;

一、数据倾斜介绍与定位

二、解决方法一&＃xff1a;聚合数据源

三、解决方法二&＃xff1a;提高shuffle操作reduce并行度

四、解决方法之三&＃xff1a;随机key实现双重聚合

五、解决方法之四&＃xff1a;将reduce join 转换为map join

六、解决方法之五&＃xff1a;sample采样倾斜key进行两次join

七、解决方法之六&＃xff1a;使用随机数以及扩容表进行join

一、数据倾斜介绍与定位

a、数据倾斜的原理

在执行shuffle操作的时候&＃xff0c;大家都知道&＃xff0c;我们之前讲解过shuffle的原理。是按照key&＃xff0c;来进行values的数据的输出、拉取和聚合的。同一个key的values&＃xff0c;一定是分配到一个reduce task进行处理的。多个key对应的values&＃xff0c;总共是90万。但是问题是&＃xff0c;可能某个key对应了88万数据&＃xff0c;key-88万values&＃xff0c;分配到一个task上去面去执行。另外两个task&＃xff0c;可能各分配到了1万数据&＃xff0c;可能是数百个key&＃xff0c;对应的1万条数据。

第一个和第二个task&＃xff0c;各分配到了1万数据&＃xff1b;那么可能1万条数据&＃xff0c;需要10分钟计算完毕&＃xff1b;第一个和第二个task&＃xff0c;可能同时在10分钟内都运行完了&＃xff1b;第三个task要88万条&＃xff0c;88 * 10 &＃61;  880分钟 &＃61; 14.5个小时&＃xff1b;

b、数据倾斜的现象&＃xff0c;有两种表现&＃xff1a;

1、你的大部分的task&＃xff0c;都执行的特别特别快&＃xff0c;刷刷刷&＃xff0c;就执行完了(你要用client模式&＃xff0c;standalone client&＃xff0c;yarn client&＃xff0c;本地机器主要一执行spark-submit脚本&＃xff0c;就会开始打印log)&＃xff0c;task175 finished&＃xff1b;剩下几个task&＃xff0c;执行的特别特别慢&＃xff0c;前面的task&＃xff0c;一般1s可以执行完5个&＃xff1b;最后发现1000个task&＃xff0c;998&＃xff0c;999 task&＃xff0c;要执行1个小时&＃xff0c;2个小时才能执行完一个task。

出现数据倾斜了

还算好的&＃xff0c;因为虽然老牛拉破车一样&＃xff0c;非常慢&＃xff0c;但是至少还能跑。

2、运行的时候&＃xff0c;其他task都刷刷刷执行完了&＃xff0c;也没什么特别的问题&＃xff1b;但是有的task&＃xff0c;就是会突然间&＃xff0c;啪&＃xff0c;报了一个OOM&＃xff0c;JVM Out Of Memory&＃xff0c;内存溢出了&＃xff0c;task failed&＃xff0c;task lost&＃xff0c;resubmitting task。反复执行几次都到了某个task就是跑不通&＃xff0c;最后就挂掉。某个task就直接OOM&＃xff0c;那么基本上也是因为数据倾斜了&＃xff0c;task分配的数量实在是太大了&＃xff01;&＃xff01;&＃xff01;所以内存放不下&＃xff0c;然后你的task每处理一条数据&＃xff0c;还要创建大量的对象。内存爆掉了。

出现数据倾斜了

这种就不太好了&＃xff0c;因为你的程序如果不去解决数据倾斜的问题&＃xff0c;压根儿就跑不出来。

c、数据倾斜定位与出现问题的位置&＃xff1a;

根据log去定位

出现数据倾斜的原因&＃xff0c;基本只可能是因为发生了shuffle操作&＃xff0c;在shuffle的过程中&＃xff0c;出现了数据倾斜的问题。因为某个&＃xff0c;或者某些key对应的数据&＃xff0c;远远的高于其他的key。

1、你在自己的程序里面找找&＃xff0c;哪些地方用了会产生shuffle的算子&＃xff0c;groupByKey、countByKey、reduceByKey、join

2、看log

log一般会报是在你的哪一行代码&＃xff0c;导致了OOM异常&＃xff1b;或者呢&＃xff0c;看log&＃xff0c;看看是执行到了第几个stage&＃xff01;&＃xff01;&＃xff01;哪一个stage&＃xff0c;task特别慢&＃xff0c;就能够自己用肉眼去对你的spark代码进行stage的划分&＃xff0c;就能够通过stage定位到你的代码&＃xff0c;哪里发生了数据倾斜。去找找&＃xff0c;代码那个地方&＃xff0c;是哪个shuffle操作。

二、解决方法一&＃xff1a;聚合数据源

聚合数据源做法一&＃xff1a;

groupByKey、reduceByKey&＃xff1b;groupByKey&＃xff0c;就是拿到每个key对应的values&＃xff1b;reduceByKey&＃xff0c;说白了&＃xff0c;就是对每个key对应的values执行一定的计算。现在这些操作&＃xff0c;比如groupByKey和reduceByKey&＃xff0c;包括之前说的join。都是在spark作业中执行的。

spark作业的数据来源&＃xff0c;通常是哪里呢&＃xff1f;90%的情况下&＃xff0c;数据来源都是hive表(hdfs&＃xff0c;大数据分布式存储系统)。hdfs上存储的大数据。hive表&＃xff0c;hive表中的数据&＃xff0c;通常是怎么出来的呢&＃xff1f;有了spark以后&＃xff0c;hive比较适合做什么事情&＃xff1f;hive就是适合做离线的&＃xff0c;晚上凌晨跑的&＃xff0c;ETL(extract transform load&＃xff0c;数据的采集、清洗、导入)&＃xff0c;hive sql&＃xff0c;去做这些事情&＃xff0c;从而去形成一个完整的hive中的数据仓库&＃xff1b;说白了&＃xff0c;数据仓库&＃xff0c;就是一堆表。spark作业的源表&＃xff0c;hive表&＃xff0c;其实通常情况下来说&＃xff0c;也是通过某些hive etl生成的。hive etl可能是晚上凌晨在那儿跑。今天跑昨天的数据。

数据倾斜&＃xff0c;某个key对应的80万数据&＃xff0c;某些key对应几百条&＃xff0c;某些key对应几十条&＃xff1b;现在&＃xff0c;咱们直接在生成hive表的hive etl中&＃xff0c;对数据进行聚合。比如按key来分组&＃xff0c;将key对应的所有的values&＃xff0c;全部用一种特殊的格式&＃xff0c;拼接到一个字符串里面去&＃xff0c;比如“key&＃61;sessionid, value: action_seq&＃61;1|user_id&＃61;1|search_keyword….”。

对key进行group&＃xff0c;在spark中&＃xff0c;拿到key&＃61;sessionid&＃xff0c;values&＃xff1b;hive etl中&＃xff0c;直接对key进行了聚合。那么也就意味着&＃xff0c;每个key就只对应一条数据。在spark中&＃xff0c;就不需要再去执行groupByKey&＃43;map这种操作了。直接对每个key对应的values字符串&＃xff0c;map操作&＃xff0c;进行你需要的操作即可。key,values串。spark中&＃xff0c;可能对这个操作&＃xff0c;就不需要执行shffule操作了&＃xff0c;也就根本不可能导致数据倾斜。

或者是&＃xff0c;对每个key在hive etl中进行聚合&＃xff0c;对所有values聚合一下&＃xff0c;不一定是拼接起来&＃xff0c;可能是直接进行计算。reduceByKey&＃xff0c;计算函数&＃xff0c;应用在hive etl中&＃xff0c;每个key的values。

聚合数据源做法二&＃xff1a;

你可能没有办法对每个key&＃xff0c;就聚合出来一条数据&＃xff1b;

那么也可以做一个妥协&＃xff1b;对每个key对应的数据&＃xff0c;10万条&＃xff1b;有好几个粒度&＃xff0c;比如10万条里面包含了几个城市、几天、几个地区的数据&＃xff0c;现在放粗粒度&＃xff1b;直接就按照城市粒度&＃xff0c;做一下聚合&＃xff0c;几个城市&＃xff0c;几天、几个地区粒度的数据&＃xff0c;都给聚合起来。比如说

city_id&＃xff0c;date&＃xff0c;area_id

select ... from ... group by city_id&＃xff0c;date&＃xff0c;area_id

尽量去聚合&＃xff0c;减少每个key对应的数量&＃xff0c;也许聚合到比较粗的粒度之后&＃xff0c;原先有10万数据量的key&＃xff0c;现在只有1万数据量。减轻数据倾斜的现象和问题。

三、解决方法二&＃xff1a;提高shuffle操作reduce并行度

如果第一种方法不适合做。那么采用第二种方法&＃xff1a;提高shuffle操作的reduce并行度

将增加reduce task的数量&＃xff0c;就可以让每个reduce task分配到更少的数据量&＃xff0c;这样的话&＃xff0c;也许就可以缓解&＃xff0c;或者甚至是基本解决掉数据倾斜的问题。

a、原理图介绍&＃xff1a;

b、提升shuffle reduce端并行度的具体操作

主要给我们所有的shuffle算子&＃xff0c;比如groupByKey、countByKey、reduceByKey。在调用的时候&＃xff0c;传入进去一个参数。一个数字。那个数字&＃xff0c;就代表了那个shuffle操作的reduce端的并行度。那么在进行shuffle操作的时候&＃xff0c;就会对应着创建指定数量的reduce task。

这样的话&＃xff0c;就可以让每个reduce task分配到更少的数据。基本可以缓解数据倾斜的问题。

比如说&＃xff0c;原本某个task分配数据特别多&＃xff0c;直接OOM&＃xff0c;内存溢出了&＃xff0c;程序没法运行&＃xff0c;直接挂掉。按照log&＃xff0c;找到发生数据倾斜的shuffle操作&＃xff0c;给它传入一个并行度数字&＃xff0c;这样的话&＃xff0c;原先那个task分配到的数据&＃xff0c;肯定会变少。就至少可以避免OOM的情况&＃xff0c;程序至少是可以跑的。

c、提升shuffle reduce并行度的缺陷

治标不治本的意思&＃xff0c;因为&＃xff0c;它没有从根本上改变数据倾斜的本质和问题。不像第一个和第二个方案(直接避免了数据倾斜的发生)。原理没有改变&＃xff0c;只是说&＃xff0c;尽可能地去缓解和减轻shuffle reduce task的数据压力&＃xff0c;以及数据倾斜的问题。

实际生产环境中的经验&＃xff1a;

1、如果最理想的情况下&＃xff0c;提升并行度以后&＃xff0c;减轻了数据倾斜的问题&＃xff0c;或者甚至可以让数据倾斜的现象忽略不计&＃xff0c;那么就最好。就不用做其他的数据倾斜解决方案了。

2、不太理想的情况下&＃xff0c;就是比如之前某个task运行特别慢&＃xff0c;要5个小时&＃xff0c;现在稍微快了一点&＃xff0c;变成了4个小时&＃xff1b;或者是原先运行到某个task&＃xff0c;直接OOM&＃xff0c;现在至少不会OOM了&＃xff0c;但是那个task运行特别慢&＃xff0c;要5个小时才能跑完。

那么&＃xff0c;如果出现第二种情况的话&＃xff0c;各位&＃xff0c;就立即放弃这种方法&＃xff0c;开始去尝试和选择后面的方法解决。

四、解决方法之三&＃xff1a;随机key实现双重聚合

 原理图介绍&＃xff1a;

使用场景&＃xff1a;(1)groupByKey(2)reduceByKey

join&＃xff0c;咱们通常不会这样来做&＃xff0c;后面有针对不同的join造成的数据倾斜的问题的解决方案。

五、解决方法之四&＃xff1a;将reduce join 转换为map join

普通reduce join&＃xff1a;                      

map join:

普通的join

肯定是要走shuffle&＃xff1b;那么&＃xff0c;所以既然是走shuffle&＃xff0c;那么普通的join&＃xff0c;就肯定是走的是reduce join。先将所有相同的key&＃xff0c;对应的values&＃xff0c;汇聚到一个task中&＃xff0c;然后再进行join。

reduce join转换为map join

如果两个RDD要进行join&＃xff0c;其中一个RDD是比较小的。一个RDD是100万数据&＃xff0c;一个RDD是1万数据。(一个RDD是1亿数据&＃xff0c;一个RDD是100万数据)其中一个RDD必须是比较小的&＃xff0c;broadcast出去那个小RDD的数据以后&＃xff0c;就会在每个executor的block manager中都驻留一份。要确保你的内存足够存放那个小RDD中的数据

这种方式下&＃xff0c;根本不会发生shuffle操作&＃xff0c;肯定也不会发生数据倾斜&＃xff1b;从根本上杜绝了join操作可能导致的数据倾斜的问题&＃xff1b;对于join中有数据倾斜的情况&＃xff0c;大家尽量第一时间先考虑这种方式&＃xff0c;效果非常好&＃xff1b;如果某个RDD比较小的情况下。

不适合的情况&＃xff1a;

两个RDD都比较大&＃xff0c;那么这个时候&＃xff0c;你去将其中一个RDD做成broadcast&＃xff0c;就很笨拙了。很可能导致内存不足。最终导致内存溢出&＃xff0c;程序挂掉。而且其中某些key(或者是某个key)&＃xff0c;还发生了数据倾斜&＃xff1b;此时可以采用最后两种方式。

特别声明&＃xff1a;

对于join这种操作&＃xff0c;不光是考虑数据倾斜的问题&＃xff1b;即使是没有数据倾斜问题&＃xff0c;也完全可以优先考虑&＃xff0c;用我们讲的这种高级的reduce join转map join的技术&＃xff0c;不要用普通的join&＃xff0c;去通过shuffle&＃xff0c;进行数据的join&＃xff1b;完全可以通过简单的map&＃xff0c;使用map join的方式&＃xff0c;牺牲一点内存资源&＃xff1b;在可行的情况下&＃xff0c;优先这么使用。不走shuffle&＃xff0c;直接走map&＃xff0c;性能肯定是提高很多的。

六、解决方法之五&＃xff1a;sample采样倾斜key进行两次join

方案的实现思路&＃xff1a;其实关键之处在于&＃xff0c;将发生数据倾斜的key&＃xff0c;单独拉出来&＃xff0c;放到一个RDD中去&＃xff1b;就用这个原本会倾斜的key RDD跟其他RDD&＃xff0c;单独去join一下&＃xff0c;这个时候&＃xff0c;key对应的数据&＃xff0c;可能就会分散到多个task中去进行join操作&＃xff0c;最后将join后的表进行union操作。

就不至于&＃xff0c;这个key跟之前其他的key混合在一个RDD中时&＃xff0c;导致一个key对应的所有数据&＃xff0c;都到一个task中去&＃xff0c;就会导致数据倾斜。

应用场景&＃xff1a;

优先对于join&＃xff0c;肯定是希望能够采用上一讲讲的&＃xff0c;reduce join转换map join。两个RDD数据都比较大&＃xff0c;那么就不要那么搞了。

针对你的RDD的数据&＃xff0c;你可以自己把它转换成一个中间表&＃xff0c;或者是直接用countByKey()的方式&＃xff0c;你可以看一下这个RDD各个key对应的数据量&＃xff1b;此时如果你发现整个RDD就一个&＃xff0c;或者少数几个key&＃xff0c;是对应的数据量特别多&＃xff1b;尽量建议&＃xff0c;比如就是一个key对应的数据量特别多。

此时可以采用咱们的这种方案&＃xff0c;单拉出来那个最多的key&＃xff1b;单独进行join&＃xff0c;尽可能地将key分散到各个task上去进行join操作。

什么时候不适用呢&＃xff1f;

如果一个RDD中&＃xff0c;导致数据倾斜的key&＃xff0c;特别多&＃xff1b;那么此时&＃xff0c;最好还是不要这样了&＃xff1b;还是使用我们最后一个方案&＃xff0c;终极的join数据倾斜的解决方案。

进一步优化&＃xff1a;

就是说&＃xff0c;咱们单拉出来了&＃xff0c;一个或者少数几个可能会产生数据倾斜的key&＃xff0c;然后还可以进行更加优化的一个操作&＃xff1b;

对于那个key&＃xff0c;从另外一个要join的表中&＃xff0c;也过滤出来一份数据&＃xff0c;比如可能就只有一条数据。userid2infoRDD&＃xff0c;一个userid key&＃xff0c;就对应一条数据。然后呢&＃xff0c;采取对那个只有一条数据的RDD&＃xff0c;进行flatMap操作&＃xff0c;打上100个随机数&＃xff0c;作为前缀&＃xff0c;返回100条数据。

单独拉出来的可能产生数据倾斜的RDD&＃xff0c;给每一条数据&＃xff0c;都打上一个100以内的随机数&＃xff0c;作为前缀。

再去进行join&＃xff0c;是不是性能就更好了。肯定可以将数据进行打散&＃xff0c;去进行join。join完以后&＃xff0c;可以执行map操作&＃xff0c;去将之前打上的随机数&＃xff0c;给去掉&＃xff0c;然后再和另外一个普通RDD join以后的结果&＃xff0c;进行union操作。

七、解决方法之六&＃xff1a;使用随机数以及扩容表进行join

当采用随机数和扩容表进行join解决数据倾斜的时候&＃xff0c;就代表着&＃xff0c;你的之前的数据倾斜的解决方案&＃xff0c;都没法使用。这个方案是没办法彻底解决数据倾斜的&＃xff0c;更多的&＃xff0c;是一种对数据倾斜的缓解。

步骤&＃xff1a;

1、选择一个RDD&＃xff0c;要用flatMap&＃xff0c;进行扩容(比较小的RDD)&＃xff0c;将每条数据&＃xff0c;映射为多条数据&＃xff0c;每个映射出来的数据&＃xff0c;都带了一个n以内的随机数&＃xff0c;通常来说&＃xff0c;会选择10以内。

2、将另外一个RDD&＃xff0c;做普通的map映射操作&＃xff0c;每条数据&＃xff0c;都打上一个10以内的随机数。

3、最后&＃xff0c;将两个处理后的RDD&＃xff0c;进行join操作。

另一个方法&＃xff1a;

sample采样倾斜key并单独进行join

1、将key&＃xff0c;从另外一个RDD中过滤出的数据&＃xff0c;可能只有一条&＃xff0c;或者几条&＃xff0c;此时&＃xff0c;咱们可以任意进行扩容&＃xff0c;扩成1000倍。

2、将从第一个RDD中拆分出来的那个倾斜key RDD&＃xff0c;打上1000以内的一个随机数。

3、join并且提供并行度。这样配合上&＃xff0c;提升shuffle reduce并行度&＃xff0c;join(rdd, 1000)。通常情况下&＃xff0c;效果还是非常不错的。打散成100份&＃xff0c;甚至1000份&＃xff0c;2000份&＃xff0c;去进行join&＃xff0c;那么就肯定没有数据倾斜的问题了吧。

此方法局限性&＃xff1a;

1、因为你的两个RDD都很大&＃xff0c;所以你没有办法去将某一个RDD扩的特别大&＃xff0c;一般咱们就是扩10倍。

2、如果就是10倍的话&＃xff0c;那么数据倾斜问题&＃xff0c;的确是只能说是缓解和减轻&＃xff0c;不能说彻底解决。

大家都在看

▼

spark sql源码系列:

是时候学习真正的spark技术了 丨 从0到1认识 spark sql 丨 spark sql 源码剖析 PushDownPredicate  丨 spark sql 源码剖析 OptimizeIn 篇

structured streaming 系列:

structured streaming 原理剖析  丨 structured streaming 碰上kafka 丨structured streaming 是如何搞定乱序时间的

spark streaming 系列:

spark streaming 读取kafka各种姿势详解  丨 spark streaming流式计算中的困境与解决之道 

spark core 系列:

彻底搞懂spark shuffle过程(1) 丨 彻底搞懂spark shuffle过程(2)  丨spark内存管理-Tungsten框架探秘

spark 机器学习系列:

学习了 streaming 和 sql&＃xff0c;别忘了还有 Mlib

关注【spark技术分享】

一起撸spark源码&＃xff0c;一起玩spark最佳实践