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分组排序内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
这篇文章主要介绍了SpringBoot静态资源路径配置及主页显示,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下2020-04-04
这篇文章主要介绍了.NET程序员如何快入门Spring Boot,微软给VS Code提供了编写Java的插件，你可以在VS Code上愉快地写Java。,需要的朋友可以参考下2019-06-06
Spring init-method与destroy-method属性的用法解析
这篇文章主要介绍了Spring init-method与destroy-method属性的用法，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2021-08-08
详解Springboot @Cacheable 注解(指定缓存位置)
这篇文章主要介绍了详解Springboot @Cacheable 注解(指定缓存位置),使用 @Cacheable 注解就可以将运行结果缓存,以后查询相同的数据,直接从缓存中取,不需要调用方法,需要的朋友可以参考下2023-09-09
这篇文章主要和小伙伴们简单介绍下 Spring Security 中的动态权限方案，以便于小伙伴们更好的理解 TienChin 项目中的权限方案，感兴趣的可以了解一下2022-06-06
SpringBoot 实现CAS Server统一登录认证的详细步骤
CAS（Central Authentication Service）中心授权服务，是一个开源项目，目的在于为Web应用系统提供一种可靠的单点登录，这篇文章主要介绍了SpringBoot 实现CAS Server统一登录认证,需要的朋友可以参考下2024-02-02
这篇文章主要介绍了Java冒泡排序简单实现，具有一定借鉴价值，需要的朋友可以参考下。2017-12-12
本文章向大家介绍Java多态数组，主要包括Java多态数组使用实例、基本知识点总结和需要注意事项，具有一定的参考价值，需要的朋友可以参考一下2022-05-05
这篇文章主要介绍了java 数据结构之栈与队列的相关资料,这里对java中的栈和队列都做出实现实例来帮助大家理解学习数据结构，需要的朋友可以参考下2017-07-07
在使用Maven进行Java项目的构建时,正确的文件编码设置对于确保项目能够正确编译和运行至关重要,本文将详细介绍如何在Maven中设置编译文件的编码,以确保项目能够正确处理各种语言的文本内容,需要的朋友可以参考下2025-02-02
最新评论