数据倾斜

前言：数据倾斜通常在shuffle时发生，因此解决数据倾斜的思路有三个：

1，把小表广播避免shuffle；

2，把大key加随机前缀打散；

3，把大key过滤出来单独处理；

• 而触发shuffle的算子可以分为三种：
  • 分组（group）：通用方法是：给大key加随机前缀，然后分两阶段聚合；还有就是map端聚合；
  • 连接（join）：
    • 大表 join小表：把小表广播，在map端 join；
    • 大表 join大表：大表a有倾斜key，而大表b的key没有倾斜时，1膨胀法：大表的倾斜key加[1,n]的前缀，小表对应的key膨胀n倍，join后再去前缀；2广播：把大表的倾斜key过滤出来，小表对应的key广播，然后再map端join，大表的非倾斜key正常join，最后union两个数据集；
    • 笛卡尔积：小表能广播就广播；由于笛卡尔积后通常都是取topN，所以可以将一个分区数据glom为一条数组数据，再做笛卡尔积，此时就是分区粒度的笛卡尔积(而不是数据粒度的笛卡尔积)，然后在 join后的分区内做数据粒度的笛卡尔积，同时截断为topN；这种方法可以减少shuffle时数据量；
  • 排序（sort）：由于spark采用 tera排序，会采用RangePartitioner，所以倾斜key会集中在一个分区中；可以给key增加一个随机列变成(key,rd)，然后使用二次排序，便可以把相同key分散到不同分区中，最后在去掉随机列；

1.0 控制分区数

• spark.default.parallelism：RDD的默认并行度（分区数）；如果没有使用coalesce、reparation，或者shuffle类算子没有指定分区数时，就会使用默认的并行度；并行度的设置非常重要，决定了是否能够充分利用分配的物理资源，也就是说，给了你充足的executor、core和内存，而你程序却没有用上；官网给出的并行度建议是executor*core的2_{3倍；经过测试，每个task处理200m}300m的数据量时，性能最优；sparksql读取文件时的默认并行度；

  • 默认值：在yarn模式下，如果没有设置该参数，则默认值是max(分配的总核数, 2)；
  • 例子：一个任务需要处理100亿条数据，每条数据约25B，共250G；按照每个task处理250M来计算，就需要1000个task，那么程序中的并行度设置为1000，则executor*core*3=1000，那么建议executor设置为100，core设置为4，平均每个core处理2.5个task；

• spark.sql.shuffle.partitions：DataFrame的shuffle默认并行度，与上面的RDD并行度类似；千万要注意，设置了RDD并行度只对RDD有效，要设置DataFrame并行度才对Dataframe有效，所以最好这两个参数都设置；

  • 默认值：200

• 默认分区数：均为yarn模式下；a.union(b)的最终分区是a+b，a与b笛卡尔积是a*b；

  // 该参数可以控制rdd一个切片的大小，如果想增大一个切片的大小，可以调大该参数；
  spark.hadoop.mapreduce.input.fileinputformat.split.minsize
  // 控制sparksql一个切片的大小
  spark.sql.files.maxPartitionBytes
  // 生成spark.context时，会生成下面两个参数，由下面两个参数可以推算出rdd的分区数；
  sc.defaultParallelism     = spark.default.parallelism（总核数）
  sc.defaultMinPartitions = min(spark.default.parallelism,2)
  

  • RDD：

    • scala集合生成（parallelize）：如果没有指定分区数，则默认是defaultParallelism；

    • 读文件时（textFile）：max(文件的分片数，defaultMinPartitions)，hdfs是128M一片，本地文件是32M；

      如果textFile设置了分区参数m，则最后的分区数为max(文件的分片数, m)；

    • shuffle时：等于defaultParallelism，没设置就选两个rdd分区最大值；

  • SparkSQL：

    • 读文件时：max(文件的分片数，defaultMinPartitions)；
    • shuffle时：默认是spark.sql.shuffle.partitions；

• coalesce(10)：不触发shuffle的重分区，以分区为最小单位进行重分区，一般在filter之后最好加上coalesce；

  • 注意：如果原本有100个分区，现在使用coalesce(10)合并为10个分区，由于没有触发shuffle，所以是一个task执行10个分区的数据，容易导致OOM，所以coalesce的分区数不要与原分区数差异太大；

• repartition(10)：触发shuffle的coalesce重分区，以数据为最小单位进行重分区，在分组和union之后最好调用；

• partitionBy(new HashPartitioner(10))：按照自己的分区规则分区，这里是按照Hash值分区；

1.1 小表 join 大表时数据倾斜

• 背景：有一个全量商品表（小表，约20万），和一个用户日志表（大表，约20亿），现在要 join 这两个表；而大表中因为一些热销品，导致热销商品的日志记录非常多，大部分商品的记录在4万 ~ 6万，而热销商品的记录将近100万，所以就导致了数据倾斜；如果小表非常小（几百兆到1g）则可以将小表广播，然后在map端join；大于1g则可以使用下面的方法；

• 解决思路：首先将大表倾斜的key过滤出来，加上[0, 20）的随机前缀；然后把小表中相应的key膨胀20倍，对每一个key加上[0,20）的前缀，然后 join；对于非倾斜的key不做加工，然后 union 倾斜key的结果和非倾斜key的结果；

  • 该方法处理的场景是：一个表中有倾斜key，而另外一个表中的key比较均匀，即使是大表join大表，只要满足这个条件就可以使用该方法；

• 实现步骤：

  • 过滤倾斜key：对大表中的key使用sample(false, 0.1)做10%的不放回抽样，然后统计每个key出现的次数，取top20，或者次数超过一定阈值的key作为倾斜key集合slant，然后将slant广播；

  • 把大表中的倾斜key加[0, n)的随机前缀：首先使用filter(x=>slant.contains(x))过滤出倾斜的key，然后使用map对大表倾斜的key加上随机前缀，例如：map(x=>(Random.nextInt(20)+"_"+x._1, x._2))，这里是加上[0, 20)的随机前缀，也就是将每一个大key拆分为20个新的key；

  • 把小表中的倾斜key膨胀n倍：同样，使用filter过滤出倾斜key，然后使用 flatMap(x=>

    val arr = new mutable.ArrayBuffer[(key, value)]

    for(i <- 0 until 20){arr.append(i+"_"+x._1, x._2) }

    arr) 将小表中的key膨胀20倍；

  • 分别 join 再 union：把大表和小表中的倾斜key join，非倾斜key与非倾斜key join，然后再 union 两者的结果，就是最终的结果了；

1.2 笛卡尔积数据倾斜

• 背景：推荐系统中经常需要计算商品或用户的两两相似度，例如基于用户的协同过滤UserCF就需要计算用户之间的两两相似度，其中需要对（item，List(user)）中的List(user)做笛卡尔积，得到（user1，user2，score）的用户相似度；但是对于热销品，List(user)中的user会非常多，例如10万个，那么笛卡尔积后就是10的10次方，这还是一个key，通常一个task会处理多个key；而对于这么大的数据量，单个task是处理不了的，导致OOM；
• 解决思路：将大key单独过滤出来，然后分别发送到 driver端，再把每一个key封装成一个RDD，然后再跟自己做笛卡尔积join，最后将所有的结果 union起来；
• 实现步骤：
  • 过滤倾斜key：根据List(user)的长度排序，然后将top20或者超过一定阈值的倾斜key过滤出来并collect到driver端；
  • 将倾斜key转为RDD：遍历collect到driver端的倾斜key，将一个key转为一个RDD，然后再跟自己做笛卡尔积 join，最后再计算相似度；

1.3 笛卡尔积分组topN问题

• 背景：现在需要对5w用户与660w用户两两计算余弦相似度，这5w个用户中，每个用户都要与660w用户计算相似度，然后得到打分top200的用户；5w * 660w = 3.3 * 10e11，数据格式为 (userid, 特征序列)，每条数据1k；如果按照一般的计算方式：先做笛卡尔积，然后计算相似度，最后根据userid分组取top200；但是在做完笛卡尔积之后，数据量膨胀到3.3 * 10e11 * 1k = 330T，数据量非常大，即使给3000并行度，每个task也需要计算110G的数据，跑了15个小时还没跑完；
• 解决思路：将5w和660w用户，每个分区的用户变成一个条数据(数组)，然后分区之间做笛卡尔积（而不是数据之间做笛卡尔积），然后每个分区遍历两个数组，求相似度并取top200；如果小表小于1G，可以直接广播然后 join；
• 实现步骤：
  • 将一个分区数据变成一条数据：kRdd：5w用户分为30个分区，平均一个分区1.7k条数据；cRdd：660w用户分为100个分区，平均一个分区6.6w条数据；然后用glom()算子将这两个RDD的每个分区变成一条数组数据；
  • 分区间做笛卡尔积：kRdd与cRdd做笛卡尔积，变成30*100=3000个分区，每个分区只有一条数据，格式为(kPart: Array[String, Double], cPart: Array[String, Double])，大小为kPart(1.7k *1kb) + cPart(6.6w *1kb)=67.7M所以总的计算数据大小为67.7M *3000=203G；
  • 分区内计算相似度求top200：分区内计算1.7k用户与6.6w用户的相似度，并取top200；每个分区的计算量是1.7k *6.6w=1 *10e8，返回的数据量是1.7k *200 *0.2kb=68M；返回的总数据量是68M *3000=204G，给1500核3000并行度，平均一个核跑2个task，跑完一个task平均需要40分钟，因此总耗时为80分钟；这一步要关注两层for循环中的代码优化，例如已知数组的最大长度时，在初始化数组时就给一个初始长度，可以减少数组扩容带来的性能消耗，性能有较高的提升；
  • 分组求全局top200：上一步返回的是RDD[(key: String, value: Array[(String, Double)])]，是每个key在分区中的top200，这一步就是把不同分区中的top200合并再求全局top200，由于这一步的数据量比较小而且没有数据倾斜问题，所以可以直接使用groupByKey算子；

3，Spark数据倾斜

原文：https://www.cnblogs.com/shendeng23/p/15240771.html