spark数据倾斜

Posted 2020-12-02 xiashiwendao

数据倾斜的主要问题在于，某个分区数量很巨大，在做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/