Spark 使用局部聚合+全局聚合解决数据倾斜问题

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

例如，reduced端一共要处理100万条数据，第一个和第二个task分别被分配到了1万条数据，计算5分钟内完成，第三个task分配到了98万数据，此时第三个task可能需要10个小时完成，这使得整个Spark作业需要10个小时才能运行完成，这就是数据倾斜所带来的后果。

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataframe, SparkSession}


object _DataSkewDemo {
    def main(args: Array[String]): Unit = {

        val spark: SparkSession = SparkSession
          .builder()
          .master("local[*]")
          .appName(" ")
          .getOrCreate()

        import spark.implicits._

        //1.自创一个简单RDD进行模拟倾斜
        val rdd: RDD[String] = spark.sparkContext.makeRDD(List("a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a,a", "b,c,d,e,f", "b,b,c,c,d,e,f,g", "a,b,a,c,f"))

        //2.将RDD转成DF
        val df: Dataframe = rdd.toDF("line")
        df.createTempView("temp")
        
        val sql =
            """
              |select substr(prefix_word,instr(prefix_word,"-")+1) w,sum(num)
              |from(
              |select prefix_word,count(1) num
              |from(
              |select concat(floor(rand()*4),"-",word) prefix_word
              |from(
              |select explode(split(line,",")) word
              |from temp) t1
              |) t2
              |group by prefix_word
              |) t3
              |group by w
              |""".stripMargin

        //4.显示数据
        spark.sql(sql).show()
        spark.stop()
    }
}