上周写了篇关于数据倾斜的案例，很多小伙伴私下在练习的时候，发现用我之前提供的数据源验证的结果有些不太对劲，我仔细一看：shit... 是我搞错了，为了能模拟出比较好的数据倾斜效果，我其实是用了一份新的数据源。

不过好在发现这个问题之后，我及时把我做验证的数据源给上传到了网盘，大家可以安心用这个数据源做验证了，截止到目前为止，我给大家提供的数据源从之前的1个大文件，增加到了4个，覆盖的数据日期也从原来的一天，覆盖到了多天，数据总量约1.5亿，后续还会持续增加(需要数据下载地址的，私信我)。

另外，项目代码的GitHub地址为：https://github.com/Anryg/internet_behavior_project

上篇文章提到给大家一个思考题，将spark默认的分区策略，根据当前业务场景的需要进行优化，解释了spark的默认分区原理(其实主流计算引擎都是一样的)，但是通过默认的分区方式，并不能很好的将目标的4个key分派到4个不同的partition中，而我们解决数据倾斜的核心目的就是要减少单个partition的数据量。

虽然我们用加盐和减盐的方式，让单个热点key的value数据量能够有效减少，有效的解决了数据倾斜的问题，但是如果我们能更进一步，让数根据不同key被不同的task处理，那么就会让数据处理进一步被平均化，效率也会进一步提升。

|如何优化

因为默认的数据分区策略为HashPartitioner，又因为哈希碰撞的问题，导致了其中两个不同的key(target_ip)，进入了同一个partition中。

那么解决办法就是根据当前的实际业务需要，重写新的Partitioner，方式其实也特别简单，就是新写一个分区类，让其继承Partitioner抽象类，自定义数据与partition id之间的关系，示例代码如下：

package com.anryg.bigdata.test.data_skew

import org.apache.spark.Partitioner

/**
  * @DESC: 实现自定义的分区策略
  * @Auther: Anryg
  * @Date: 2022/10/13 09:52
  */
class MyPartitioner(partitionNum: Int) extends Partitioner{
    override def numPartitions: Int = partitionNum  //确定总分区数量

    override def getPartition(key: Any): Int = {//确定数据进入分区的具体策略
        val keyStr = key.toString
        val keyTag = keyStr.substring(keyStr.length - 1, keyStr.length)
        keyTag.toInt % partitionNum
    }
}