任務(wù)需求：已知RDD[(query:String, item_id:String, imp:Int, clk:Int)]，要求找到每個(gè)query對(duì)應(yīng)的點(diǎn)擊最多的前2個(gè)item_id，即：按照query分組，并按照clk降序排序，每組取前兩個(gè)。

例如：

（連衣裙，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）

先看一個(gè)錯(cuò)誤的版本：

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)])]，根據(jù)query分組再map，line._2.toArray把Iterable轉(zhuǎn)為Array，sortBy(_._3)是按最后一個(gè)Int即clk排序，(Ordering[Int].reverse)表示從大到?。╯ortBy默認(rèn)從小到大，注意這里的sortBy是Array的成員函數(shù)而不是rdd的sortBy，用法比較不同），take(2)是取前2個(gè)，然后返回（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行是直接打印跟預(yù)期稍有差別，同一個(gè)key下的top兩個(gè)元素是作為一個(gè)整體，但已經(jīng)很接近目標(biāo)，如果希望拆分，需要使用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，那每一個(gè)key對(duì)應(yīng)的一個(gè)Iterable變量，相當(dāng)于一條數(shù)據(jù)，map后的結(jié)果自然還是一條。但flatMap，相當(dāng)于map+flat操作，這才是我們真正的需要的形式。

任務(wù)進(jìn)階：要求找到每個(gè)query對(duì)應(yīng)的點(diǎn)擊最多的前2個(gè)item_id，當(dāng)點(diǎn)擊一樣時(shí)，選曝光最少的，即：按照query分組，并優(yōu)先按照clk降序排序，其次按照imp升序排序，每組取前兩個(gè)。

例如：

（連衣裙，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）

注意，上面樣本中牛仔褲有兩個(gè)樣本的點(diǎn)擊都是7，但標(biāo)紅的樣本曝光數(shù)是更小，所以應(yīng)該入選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可以根據(jù)需要增加排序維度，參數(shù)按優(yōu)先級(jí)排列，這個(gè)在日常使用較多。

到此這篇關(guān)于spark中使用groupByKey進(jìn)行分組排序的文章就介紹到這了,更多相關(guān)spark使用groupByKey分組排序內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論