数据倾斜的主要问题在于,某个分区数量很巨大,在做map运算的时候,将会发生别的分区task很快计算完成,但是某几个分区task的计算成为了系统的瓶颈,明显超过其他分区时间;
1.方案:Kafka的随机主题
如果kafka的topic和分区关联,而且kafka是专用的,那么其实kafka如果能够和随机主机,那么数据将会随机打入到各个分区中,这样可以解决数据热点问题;
2. 方案:将不可切割的文件转换为可切割文件
对于gzip这类文件最好转化为可切割文件;因为对于不可切割的压缩文件,将会作为一个单独分区来处理,比如8.5G压缩到了23.5M,但是因为数据量巨大,这个gzip文件所在的分区将会成为性能瓶颈。
3. 方案:调整并行度
大量的key分配到同一个分区/Task里面;spark在数据分区的时候,默认使用的HashPartition;通过增加/减少并行度,来对于热点key进行重新分区;
另外一种方式类似:自定义分区函数,将HashPartition中集中分配的key,通过自定义的方式进行分散处理;
4. 方案:Reduce侧的Join通过BroadCast转换为Map侧的Join
select... from t1 join t2 on t1.aa=t2.bb
这个sql语句spark正常的执行方式是t1表中map,t2表中数据map(t2表数据量表较小),然后reduce到一起进行关联操作;因为t1表返回的粒度和t2表返回数据粒度不一样,所以导致reduce侧分区可能出现倾斜;
对于这场场景处理,通常都是将t2表中的数据通过broadcast的方式广播到各个分区中;这样各个分区在map阶段就可以进行数据关联,避免了shuffle到reduce阶段再进行关联。
5. 方案:拆分热点key
如果倾斜发生在一张表A中(另外一张表数据B比较均匀),而且数据热点集中在少数几个key;
1. 把大表中这几个key单独拎出来:skewRDD,在拎出来的时候,对于key添加随机数,保证对这些(添加了前缀的)key进行合理分区;然后skewRDD和B表进行join;skewJoinedRDD;
2. 过滤A表中那几个key,形成unskewRDD,和B表做join,形成unskewJoinedRDD;
3. skewJoinedRDD和unskewJoinedRDD做union(联合);
6. 方案:大表key全部进行添加随机数,小表扩大N倍
如果热点key太多了,那么就需要:
1. 把大表A所有的key都添加随机值,这样,对于大表数据进行数据重新分区,比如48取随机数,填充到key的前面,通过“,”分割;
2. 对于小表B里面所有的数据都扩充N倍,比如数据1,现在有48条数据1,48条记录value都是一样的,但是key为“1,key”,“2,key“... "48, key"
3. 然后A,B做join,这样小表B中必然有一条会和A表关联上(其余47条记录成为冗余记录)
4. A,Bjoin结果做mapToPair,去掉之前前缀,结果集OK;
看了这么多的方案,你会发现其实本质就是发现了数据倾斜,考虑怎么将热点数据进行分散(重新分区);这里可以考虑kafka的随机分区;调整并行度;把倾斜的key对应数据单独拎出来进行分区,最后和不包含这些key的key做union;全体key添加随机数,导致全体数据重新分区,再对具有N倍冗余数据的小表进行join获取数据;
比较特殊两个方案一个是对于不可分割的数据的处理,尽量转化为可分割数据(比如压缩文件搞成非压缩文件);另外一个是将join的表(小表)通过broadcast将数据进行广播,reduce阶段的join转化为map阶段的join。
参考(我觉得作者有些神人之感)
http://www.jasongj.com/spark/skew/
原文地址:https://www.cnblogs.com/xiashiwendao/p/9278778.html