【Spark】Spark数据倾斜解决方案、大表join小表及大表join大表优化思路

如果觉得这篇文章对您有帮助，别忘了点赞、分享或关注哦！您的一点小小支持，不仅能帮助更多人找到有价值的内容，还能鼓励我持续分享更多精彩的技术文章。感谢您的支持，让我们一起在技术的世界中不断进步！

Spark 数据倾斜处理

Spark 中的数据倾斜问题主要指 shuffle 过程中出现的数据倾斜问题，是由于不同的 key 对应的数据量不同导致的不同 task 所处理的数据量不同的问题。

1. 解决方案一：两阶段聚合

• 当使用了类似reduceByKey、groupByKey这样的聚合类算子时，可以考虑量两阶段聚合。
• 实现思路：第一次是局部聚合，通过map算子给每个数据的key添加随机数前缀，对key进行打散，执行reduceByKey等聚合操作，第二次将各个key的随机前缀去掉，再次进行全局聚合操作，就可以得到最终结果了。
• 对于聚合类的shuffle操作导致的数据倾斜，效果是非常不错的，通常可以解决数据倾斜问题。
• 适用范围较窄，仅仅适用聚合类的shuffle操作，像join操作还得考虑其他方案。

2. 解决方案二：广播join

• 广播Join是将小表的数据广播到每个计算节点，而不是将两个大表的数据进行Shuffle。当其中一个表的数据量较小（可以放入内存，默认值为10M，可调整），就可以使用广播连接来避免大规模的Shuffle，从而提高性能，减少数据倾斜的发生。
• 适用小表与大表连接，且小表的大小适合广播到每个Executer节点。

3. 解决方案三：使用随机数前缀和扩容RDD进行join

• 如果在进行join操作时，RDD中有大量的key导致数据倾斜，我们可以考虑对其中一个RDD数据进行扩容，另一个RDD进行稀释后再join。
• 实现思路：选择一个RDD，对每条数据的key添加随机前缀(1_{n),选择另外一个RDD，对每条数据都打上(1}n)随机前缀，也就是扩容n倍，将两个处理后的RDD进行join操作。
• 如果两个RDD都很大,扩容方法不在适用，扩容方式只能缓解数据倾斜。
• 如果有少数key导致数据倾斜，可以将这几个key从原来join的两个RDD中分别拆分出来形成一个单独的RDD，像上述思路一样使用随机数前缀和扩容RDD进行join，另外去除倾斜key的两个RDD正常join，将两次结果使用union算子合并起来即可。

4. 解决方案四：提高shuffle read task并行度

• reduce端的并行度的提高就增加了reduce端task的数量，那么每个task分配到的数据量就会相应的减少，在一定程度上可以缓解数据倾斜问题。
• 设置参数：spark.sql.shuffle.partitions ，默认值为200，适当提高。
• 举例：如果原本有4个key，每个key对应20条数据，这4个key分配给一个task，那么这个task就要处理80条数据，如果增加shuffle
  read task为4,每个task就会分配到一个key，也就是每个task处理10条数据，执行时间会有所减少。
• 该方案无法彻底解决数据倾斜，比如某个key对应的数据量有100w，无论task数量怎么增加，这个对应着100w数据的key肯定会分配到一个task中处理，还是会发生数据倾斜。
• 适用有较多个key对应的数据量都比较大的情况。

Spark大表join小表(广播变量容不下小表)

• 小表是否可以通过过滤减小存储空间，以完成广播。

• 小表是否可以通过构建联合key替换超长的join keys来减小存储空间，以完成广播。

• 若小表分布均匀，无法广播，则使用 join hints强行要求Spark SQL在运行时选择SHJ关联策略，相比SMJ，SHJ的执行效率会更好一些。原因是小表构建哈希表的开销，要小于两张表排序的开销。

  SELECT /*+ shuffle_hash(ad_click) */
         ad_click.idea_id, 
         SUM(ad_view.view_dsp) AS view_tatal
  FROM ad_view
  INNER JOIN ad_click ON ad_view.idea_id = ad_click.idea_id
  GROUP BY 
      ad_click.idea_id
  order by view_tatal desc
  

Spark大表join大表

• 分而治之，能否将大表join大表转化为大表join小表，将大表均匀拆分后放进广播变量。
• 拆分的关键在于拆分列的选取，拆分的基数要足够大才行，比如按照日期拆分多个小表。
• 可以通过遍历日期，完成大表与每个小表的join，然后结果union到一起，替代大表join大表的操作。
• 如果无法做到均匀拆分，只能依赖shuffle join去完成大表join大表，我们可以采用调优方法，例如存在数据倾斜时，使用两阶段shuffle消除倾斜，平衡Executor之间的计算负载。