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可以根據需要增加排序維度,參數按優先級排列,這個在日常使用較多。
到此這篇關於spark中使用groupByKey進行分組排序的文章就介紹到這瞭,更多相關spark使用groupByKey分組排序內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- python中列表的常見操作梳理總結(一)
- python3中dict.keys().sort()用不瞭的解決方法
- Spark-Sql入門程序示例詳解
- 淺談JS數組內置遍歷方法有哪些和區別
- python內置數據類型使用方法和繼承關系