spark分组排序二

• 前言
• 源码
• 总结

在sarpk分组排序一一文中最后提到，此方案可能会导致内存溢出。因此，写了一个优化的demo，先上代码

import org.apache.spark.{Partitioner, SparkConf, SparkContext}

object Demo2 {
  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf().setMaster("local[*]").setAppName("demo2"))

    val rdd1 = sc.textFile("./data/sku")

    // 2. 转换数据格式,字符串变元组: cate,sku => ((cate,sku), 1)
    val rdd2 = rdd1.map(line=>{
      ((line.split(",")(0), line.split(",")(1)), 1)
    })

    // 3. 统计每个sku的销量
    val rdd3 = rdd2.reduceByKey(_ + _)

    // 4. 统计有多少个品类
    val arr = rdd3.keys.map(_._1).distinct().collect()
    // 5. 创建自定义分区器，具体逻辑看类定义中注释
    val myPartitioner = new MyPartitioner(arr)

    // 6. 按照类目id将数据重分区, 每个类目的数据重新分配到一个分区
    val rdd4 = rdd3.partitionBy(myPartitioner)

    // 7. 计算每个分区内的top1：即每个品类小销量最多的sku
    val rdd5 = rdd4.mapPartitions(iter => {
      // 新建一个容量为1的数组，存储销量最好的sku信息
      val arr = new Array[((String, String), Int)](1)

      // 迭代分区内所有的数据，找出销量最大的sku
      // 这个地方就是较初版优化的地方，使用迭代器就不会将所有数据一次性加载到内存
      iter.foreach(x => {
        if(arr(0) == null) {
          arr(0) = x
        }else {
          if(arr(0)._2 < x._2) {
            arr(0) = x
          }
        }
      })
      arr.toIterator
    })

    rdd5.foreach(println)
  }
}

自定义分区器

class MyPartitioner(cates: Array[String]) extends Partitioner {

  // 分区数即为品类数
  override def numPartitions: Int = cates.length

  // 每条数据所映射到的分区为类目id的索引值
  override def getPartition(key: Any): Int = {
    val cate = key.asInstanceOf[(String, String)]
    cates.indexOf(cate._1)
  }
}

较于spark分组排序一的方案，主要优化点在于取分组内top1的逻辑，先看之前的代码逻辑，暂且叫方案一

// 5. 将每个组的元素按照倒序排序，取出第一个元素
    val rdd5 = rdd4.flatMapValues( _.toList.sortBy(-_._2).take(1))

这是新逻辑，暂且叫方案二

    // 7. 计算每个分区内的top1：即每个品类小销量最多的sku
    val rdd5 = rdd4.mapPartitions(iter => {
      // 新建一个容量为1的数组，存储销量最好的sku信息
      val arr = new Array[((String, String), Int)](1)

      // 迭代分区内所有的数据，找出销量最大的sku
      // 这个地方就是较初版优化的地方，使用迭代器就不会将所有数据一次性加载到内存
      iter.foreach(x => {
        if(arr(0) == null) {
          arr(0) = x
        }else {
          if(arr(0)._2 < x._2) {
            arr(0) = x
          }
        }
      })
      arr.toIterator
    })

主要区别在于：方案一将分类下的所有sku全部装在到内存(toList)，然后排序(sortby)，最后取top1；而方案二先是定义了一个分区器，是每个分区只存储一个品类的数据，然后在分区内使用迭代器，取出销量最大的sku。这样就不用将数据全部装在到内存

当然方案二也有几个问题
逻辑复杂，特别是自定义分区器，然后数据重新partition
取top1的逻辑是自定义的，如果改成top5、top10。。。就没法用了，这个地方可以继续优化成更通用的逻辑：用有界优先队列取topn

继续优化ing