恩里克-Henrique 智维平台

经过分析，我发现以下原因导致数据倾斜：

1. 空值集中：约22.4%的记录中用户ID为NULL，这些记录被分到了同一个分区（分区9），导致该分区数据量远高于其他分区。
2. 任务不平衡：分区9的执行任务耗时是其他分区的5-8倍，成为了整个作业的瓶颈。
3. Shuffle放大：由于分区不均衡，导致Shuffle阶段数据传输量异常增大，从1.2TB扩大到1.4TB。

这种情况下，单个Executor需要处理大量数据，而其他Executor则处于相对空闲状态，严重影响了并行计算效率。

针对这个数据倾斜问题，我建议以下优化方案：

1. 空值处理：

   // 在用户行为分析阶段，为NULL值添加随机前缀
   val randomPrefixes = (1 to 10).map(i => s"rand_$i")
   val betterDistributedRDD = userBehaviorRDD.map { record =>
     val userId = record.userId
     if (userId == null) {
       val randomPrefix = randomPrefixes(Random.nextInt(10))
       record.copy(userId = s"${randomPrefix}_null")
     } else {
       record
     }
   }

2. 自定义分区策略：

   // 实现自定义分区器，确保NULL值均匀分布
   class CustomPartitioner(partitions: Int) extends Partitioner {
     def numPartitions: Int = partitions
     def getPartition(key: Any): Int = {
       val k = key.toString
       if (k.contains("rand_")) {
         // 从前缀提取数字并用作分区索引
         val prefix = k.substring(5, k.indexOf("_null"))
         prefix.toInt % numPartitions
       } else {
         Math.abs(k.hashCode % numPartitions)
       }
     }
   }

3. 增加并行度：

   // 增加用户行为分析阶段的并行度
   spark.conf.set("spark.sql.shuffle.partitions", 200)
   // 或者在RDD操作中指定
   userBehaviorRDD.repartition(200)

4. 预聚合优化：

   // 在shuffle前进行本地预聚合，减少数据传输量
   userBehaviorRDD
     .map(record => (record.userId, 1))
     .reduceByKey((a, b) => a + b, 50) // 本地预聚合，50个分区
     .repartition(200)
     .reduceByKey((a, b) => a + b) // 全局聚合

应用这些优化后，预计可将第4阶段执行时间从42分钟降至8-10分钟，整体作业时间可缩短约70%。