大数据计算时数据倾斜问题及解决方案

首先谈一下什么是数据倾斜? 

map /reduce 程序执行时，reduce 节点大部分执行完毕，但是有一个或者几个 reduce 节点运行很慢，导致整个程序的处理时间很长。现象是: 进度长时间维持在 99%（或 100%），查看任务监控页面，发现只有少量（1 个或几个）reduce 子任务未完成；查看未完成的子任务，可以看到本地读写数据量积累非常大，通常超过 10GB 可以认定为发生数据倾斜。

如何发生的?数据的倾斜主要是两个的数据相差的数量不在一个级别上 ,这是因为某一个 key 的条数比其他 key 多很多（有时是百倍或者千倍之多）,这条 key 所在的 reduce 节点所处理的数据量比其他节点就大很多，从而导致某几个节点迟迟运行不完。

如何定位发生数据倾斜的代码 

1、数据倾斜只会发生在 shuffle 中，下面是常用的可能会触发 shuffle 操作的算子：distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition 等。出现数据倾斜时，可能就是代码中使用了这些算子的原因 

2、通过观察 spark UI 的节目定位数据倾斜发生在第几个 stage 中，如果是用 yarn-client 模式提交，那么本地是可以直接看到 log 的，可以在 log 中找到当前运行到了第几个 stage;如果用 yarn-cluster 模式提交，可以通过 Spark Web UI 来查看当前运行到了第几个 stage。此外，无论是使用了 yarn-client 模式还是 yarn-cluster 模式，我们都可以在 Spark Web UI 上深入看一下当前这个 stage 各个 task 分配的数据量，从而进一步确定是不是 task 分配的数据不均匀导致了数据倾斜。 

3、根据之前学的 stage 的划分算法定位到极有可能发生数据倾斜的代码

查看导致数据倾斜的 key 的分布情况 

1. 如果是 Spark SQL 中的 group by、join 语句导致的数据倾斜，那么就查询一下 SQL 中使用的表的 key 分布情况。 

2. 如果是对 Spark RDD 执行 shuffle 算子导致的数据倾斜，那么可以在 Spark 作业中加入查看 key 分布的代码，比如 RDD.countByKey()。然后对统计出来的各个 key 出现的次数，collect/take 到客户端打印一下，就可以看到 key 的分布情况。

怎么解决数据倾斜

优化方案 : 

方式一 : reduce 本身的计算需要以合适的内存作为支持，在硬件环境容许的情况下，增加 reduce 的 JVM 内存大小显然有改善数据倾斜的可能，这种方式尤其适合数据分布第一种情况，单个值有大量记录, 这种值的所有纪录已经超过了分配给 reduce 的内存，无论你怎么样分区这种情况都不会改变. 当然这种情况的限制也非常明显, 1.内存的限制存在，2.可能会对集群其他任务的运行产生不稳定的影响。

方式二 : 这个对于数据分布第二种情况有效，情况(一值较多，单个唯一值的记录数不会超过分配给 reduce 的内存). 如果发生了偶尔的数据倾斜情况，增加 reduce 个数可以缓解偶然情况下的某些 reduce 不小心分配了多个较多记录数的情况. 但是对于第一种数据分布无效。

方式三: 一种情况是某个领域知识告诉你数据分布的显著类型，比如<<hadoop 权威指南>> 里面的温度问题，一个固定的组合(观测站点的位置和温度) 的分布是固定的, 对于特定的查询如果前面两种方式都没用，实现自己的 partitioner 也许是一个好的方式。

总结: 数据倾斜没有一劳永逸的方式可以解决，了解你的数据集的分布情况，然后了解你所使用计算框架的运行机制和瓶颈，针对特定的情况做特定的优化，做多种尝试，观察是否有效。