Спарк topN значения по группам

Question

У меня есть фрейм данных с подробной информацией о событии, я пытаюсь получить топ-5 недавно зарегистрированных событий по дате и идентификатору пользователя. Вот код, который я пробовал до сих пор.

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|
+-------+-------+--------------+

... вот часть, с которой я борюсь, как соединить, объединить последнюю группу, о которой сообщают, чтобы ранжировать и получить лучшие 5 из последней сообщенной группы. ...

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)

Также, как получить 5 лучших отфильтрованных рангов, в этом случае у нас может быть несколько событий с одинаковым рангом.

+--------------+-------+-------+-----------+--------------------------+----+
|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   |
+--------------+-------+-------+-----------+--------------------------+----+

Answer 1

Я остановился на этом решении. Сначала объедините данные по последнему сообщенному времени, а затем объедините их с исходным DF, чтобы исключить все нежелательные данные и выполнить ранжирование для полученных данных.

     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   |
+--------------+-------+-------+-----------+----------------------------+----+