spark中如何使用groupByKey进行分组排序

作者:有用网 阅读量:400 发布时间:2023-07-12
关键字 spark

今天小编给大家分享一下spark中如何使用groupByKey进行分组排序的相关知识点,内容详细,逻辑清晰,相信大部分人都还太了解这方面的知识,所以分享这篇文章给大家参考一下,希望大家阅读完这篇文章后有所收获,下面我们一起来了解一下吧。

任务需求:已知RDD[(query:String, item_id:String, imp:Int, clk:Int)],要求找到每个query对应的点击最多的前2个item_id,即:按照query分组,并按照clk降序排序,每组取前两个。

例如:

(连衣裙,1234,  22,  13)

(牛仔裤,2768,  34,  7)

(连衣裙,1673,45,  9)

(衬衣,3468, 67,  12)

(牛仔裤,2754, 68, 20)

(连衣裙,1976,93,  29)

希望得到:

(连衣裙,1976,93,  29)

(连衣裙,1234,  22,  13)

(牛仔裤,2754, 68, 20)

(牛仔裤,2768,  34,  7)

(衬衣,3468, 67,  12)

先看一个错误的版本:

val list = List(("连衣裙",1234, 22, 13),("牛仔裤",2768, 34, 7),("连衣裙",1673,45, 9)
    ,("衬衣",3468,67, 12),("牛仔裤",2754, 68, 20),("连衣裙",1976,93, 29))
val rdd = ss.sparkContext.parallelize(list)
 
val topItem_set= rdd.map(ele => (ele._1, (ele._2, ele._3, ele._4))).groupByKey()
  .map(line => {
        val topItem = line._2.toArray.sortBy(_._3)(Ordering[Int].reverse).take(2)
        topItem.mkString(",")
        topItem.map(x => {(line._1, x._1, x._2, x._3)})
  })
topItem_set.foreach(println)
println()
topItem_set.map(_.mkString).foreach(println)

我们把query作为key,其余放到一起,groupByKey后(map之前),类型为:RDD[(String, Iterable[(String, Int, Int)])],根据query分组再map,line._2.toArray把Iterable转为Array,sortBy(_._3)是按最后一个Int即clk排序,(Ordering[Int].reverse)表示从大到小(sortBy默认从小到大,注意这里的sortBy是Array的成员函数而不是rdd的sortBy,用法比较不同),take(2)是取前2个,然后返回(query,  item_id)。跑一下上面的过程。

返回:

[Lscala.Tuple4;@2b672e4
[Lscala.Tuple4;@52e50126
[Lscala.Tuple4;@1362b124
 
(连衣裙,1976,93,29)(连衣裙,1234,22,13)
(衬衣,3468,67,12)
(牛仔裤,2754,68,20)(牛仔裤,2768,34,7)

上面3行是直接打印跟预期稍有差别,同一个key下的top两个元素是作为一个整体,但已经很接近目标,如果希望拆分,需要使用flatMap:

val topItem_set= rdd.map(ele => (ele._1, (ele._2, ele._3, ele._4))).groupByKey()
  .flatMap(line => {
        val topItem = line._2.toArray.sortBy(_._3)(Ordering[Int].reverse).take(2)
        topItem.mkString(",")
        topItem.map(x => {(line._1, x._1, x._2, x._3)})
  })

为什么呢?GroupByKey后,类型为RDD[(String, Iterable[(String, Int, Int)])],如果用map,那每一个key对应的一个Iterable变量,相当于一条数据,map后的结果自然还是一条。但flatMap,相当于map+flat操作,这才是我们真正的需要的形式。

任务进阶:要求找到每个query对应的点击最多的前2个item_id,当点击一样时,选曝光最少的,即:按照query分组,并优先按照clk降序排序,其次按照imp升序排序,每组取前两个。

例如:

(连衣裙,1234,  22,  13)

(牛仔裤,2768,  34,  7)

(连衣裙,1673,45,  9)

(衬衣,3468, 67,  12)

(牛仔裤,2754, 68, 20)

(连衣裙,1976,93,  29)

(牛仔裤,1232,  20, 7)

希望得到:

(连衣裙,1976,93,  29)

(连衣裙,1234,  22,  13)

(牛仔裤,2754, 68, 20)

(牛仔裤,1232,  20,  7)

(衬衣,2768,  34,  7)

注意,上面样本中牛仔裤有两个样本的点击都是7,但标红的样本曝光数是更小,所以应该入选top2,直接上代码吧:

val list2 = List(("连衣裙",1234, 22, 13),("牛仔裤",2768, 34, 7),("连衣裙",1673,45, 9)
    ,("衬衣",3468,67, 12),("牛仔裤",2754, 68, 20),("连衣裙",1976,93, 29),("牛仔裤",1232, 20, 7))
    val rdd2 = ss.sparkContext.parallelize(list2)
    rdd2.foreach(println)
    val topItem_set= rdd2.map(ele => (ele._1, (ele._2, ele._3, ele._4))).groupByKey()
      .flatMap(line => {
        val topItem = line._2.toArray.sortBy(x => (x._3, x._2))(Ordering.Tuple2(Ordering[Int].reverse, Ordering[Int])).take(2)
        topItem.map(x => {(line._1, x._1, x._2, x._3)})
      })
    topItem_set.foreach(println)

sortBy可以根据需要增加排序维度,参数按优先级排列,这个在日常使用较多。


#发表评论
提交评论