Spark topN values by group

Question

I have a data frame with event details, I am trying to get top 5 recently reported event by date and userid. Here the code I have tried so far.

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
val df = sc.parallelize(Seq( ("20180515114049", "user001","e001","cross-over","some data related to even"),
  ("20180515114049", "user004","e002","cross-limit","some data related to event"),
  ("20180515114049", "user001","e001","cross-over","some data related to event"),
  ("20180615114049", "user001","e001","cross-over","some data related to event"),
  ("20180715114049", "user003","e004","cross-cl","some data related to event"),
  ("20180715114049", "user005","e001","cross-over","some data related to event"),
  ("20180715114049", "user005","e002","cross-limit","some data related to event"),
   ("20180715114049", "user005","e003","no-cross","some data related to event"),
   ("20180715114049", "user005","e004","cross-over","some data related to event"),
   ("20180715114049", "user005","e005","dl-over","some data related to event"),
   ("20180715114049", "user005","e003","no-cross","some data related to event"),
  ("20180815114049", "user006","e001","cross-over","some data related to event"),
  ("20180915114049", "user001","e001","cross-over","some data related to event"),
  ("20180105114049", "user001","e006","straight","some data related to event")
 )).toDF("eventtime", "userid","eventid","event_title","eventdata")
df.show()
+--------------+-------+-------+-----------+--------------------+
|     eventtime| userid|eventid|event_title|           eventdata|
+--------------+-------+-------+-----------+--------------------+
|20180515114049|user001|   e001| cross-over|some data related...|
|20180515114049|user004|   e002|cross-limit|some data related...|
|20180515114049|user001|   e001| cross-over|some data related...|
|20180615114049|user001|   e001| cross-over|some data related...|
|20180715114049|user003|   e004|   cross-cl|some data related...|
|20180715114049|user005|   e001| cross-over|some data related...|
|20180715114049|user005|   e002|cross-limit|some data related...|
|20180715114049|user005|   e003|   no-cross|some data related...|
|20180715114049|user005|   e004| cross-over|some data related...|
|20180715114049|user005|   e005|    dl-over|some data related...|
|20180715114049|user005|   e003|   no-cross|some data related...|
|20180815114049|user006|   e001| cross-over|some data related...|
|20180915114049|user001|   e001| cross-over|some data related...|
|20180105114049|user001|   e006|   straight|some data related...|
+--------------+-------+-------+-----------+--------------------+
val df2= df.groupBy($"userid",$"eventid").agg(last($"eventtime") as "lasteventtime")
df2.show(false)
+-------+-------+--------------+
|userid |eventid|lasteventtime |
+-------+-------+--------------+
|user005|e004   |20180715114049|
|user005|e001   |20180715114049|
|user001|e006   |20180105114049|
|user001|e001   |20180915114049|
|user005|e002   |20180715114049|
|user006|e001   |20180815114049|
|user004|e002   |20180515114049|
|user005|e005   |20180715114049|
|user005|e003   |20180715114049|
|user003|e004   |20180715114049|
+-------+-------+--------------+

... here is the part I am struggling how to connect, aggregated last reported group to rank and get the top 5 of the last reported. ...

val w = Window.partitionBy($"userid",$"event_title",$"eventid").orderBy($"eventtime".desc)
val contentByRank = df.withColumn("rank", dense_rank().over(w)).filter($"rank" <= 5)
contentByRank.show(20,false)

Also how to get the top 5 of the filtered rank, in this case we may have multiple events with the same rank.

+--------------+-------+-------+-----------+--------------------------+----+
|eventtime     |userid |eventid|event_title|eventdata                 |rank|
+--------------+-------+-------+-----------+--------------------------+----+
|20180515114049|user004|e002   |cross-limit|some data related to event|1   |
|20180715114049|user005|e004   |cross-over |some data related to event|1   |
|20180815114049|user006|e001   |cross-over |some data related to event|1   |
|20180715114049|user005|e003   |no-cross   |some data related to event|1   |
|20180715114049|user005|e003   |no-cross   |some data related to event|1   |
|20180715114049|user005|e005   |dl-over    |some data related to event|1   |
|20180715114049|user003|e004   |cross-cl   |some data related to event|1   |
|20180715114049|user005|e001   |cross-over |some data related to event|1   |
|20180105114049|user001|e006   |straight   |some data related to event|1   |
|20180715114049|user005|e002   |cross-limit|some data related to event|1   |
|20180915114049|user001|e001   |cross-over |some data related to event|1   |
|20180615114049|user001|e001   |cross-over |some data related to event|2   |
|20180515114049|user001|e001   |cross-over |some data related to even |3   |
|20180515114049|user001|e001   |cross-over |some data related to event|3   |
+--------------+-------+-------+-----------+--------------------------+----+

Look here https://stackoverflow.com/questions/46058144/find-the-top-n-elements-for-attribute-combination-in-data-frame-spark — thebluephantom, Jul 03 '18 at 23:19

score 0 · Accepted Answer · answered Jul 04 '18 at 14:19

I have settled on this solution. First aggregate the data by last reported time and then join it with the original DF to eliminate all unwanted data and run a rank on the resulting data.

     val df2= df.groupBy($"userid",$"eventid").agg(last($"eventtime") as "eventtime")
     val lasteventdf=df.join(df2,Seq("eventtime", "userid","eventid"))       
     val w = Window.partitionBy($"userid",$"event_title",$"eventid").orderBy($"eventtime".desc)
     val contentByRank = lasteventdf.withColumn("rank", dense_rank().over(w)).filter($"rank" <= 5)
     contentByRank.show(20,false)

--------------+-------+-------+-----------+----------------------------+----+
|eventtime     |userid |eventid|event_title|eventdata                   |rank|
+--------------+-------+-------+-----------+----------------------------+----+
|20180515114049|user004|e002   |cross-limit|some data related to event  |1   |
|20180715114049|user005|e004   |cross-over |some data relat7ed to event |1   |
|20180815114049|user006|e001   |cross-over |some data re22lated to event|1   |
|20180715114049|user005|e003   |no-cross   |some data relate6d to event |1   |
|20180715114049|user005|e003   |no-cross   |some data rel9ated to event |1   |
|20180715114049|user005|e005   |dl-over    |some data relat8ed to event |1   |
|20180715114049|user003|e004   |cross-cl   |some data related2 to event |1   |
|20180715114049|user005|e001   |cross-over |some data related4 to event |1   |
|20180105114049|user001|e006   |straight   |some data relat4ed to event |1   |
|20180715114049|user005|e002   |cross-limit|some data related5 to event |1   |
|20180915114049|user001|e001   |cross-over |some data rel3ated to event |1   |
+--------------+-------+-------+-----------+----------------------------+----+

Spark topN values by group

1 Answers1