在火花窗口函数中聚合多列

Question

我正在尝试聚合多个列。这是我的场景。

我有这样的列 TagID、ListnerID、Timestamp 和 RSSI_weightage。

我想为每个 TagID 选择最高的 rssi_weightage。而且我还需要为上述关联选择最高的时间戳。

示例。

TagID,ListnerID,Timestamp,RSSI_Weightage
2,101,1496745906,90
3,102,1496745907,70
3,104,1496745906,80
2,101,1496745909,60
4,106,1496745908,60

My expected output

2,101,1496745909,90
3,104,1496745907,80
4,106,1496745908,60

解释

对于 TagID 2，最高的 rssi_weighatge 是 90。但最新的时间戳是 1496745909。这是我的第一行。我对所有 tagID 都做了同样的事情。

以上是我的预期输出。我试过这样。

val window = Window.partitionBy("tagShortID", "ListenerShortID").orderBy("RSSI_Weightage","Timestamp")
 val prox = averageDF.withColumn("rank", row_number().over(window)).where($"rank" === 1) 

但是上面的语句不能正常工作。我得到不正确的结果。此外，如果我按顺序应用 desc，则会引发错误。

Answer 1

给定数据框

+----------+---------------+----------+---------------+
|tagShortID|ListenerShortID|Timestamp |RSSI_Weight_avg|
+----------+---------------+----------+---------------+
|2         |101            |1496745906|90             |
|3         |102            |1496745907|70             |
|3         |104            |1496745906|80             |
|2         |101            |1496745909|60             |
|4         |106            |1496745908|60             |
+----------+---------------+----------+---------------+

您可以使用Window 函数执行以下操作。但是你将不得不做一些额外的任务，比如filter 和drop

import org.apache.spark.sql.functions._
val window = Window.partitionBy($"tagShortID")
val prox = averageDF.withColumn("RSSI_Weight_avg", max($"RSSI_Weight_avg").over(window))
  .withColumn("rank", rank().over(window.orderBy($"Timestamp".desc)))
  .filter($"rank" === 1)
  .drop("rank")

你会得到结果

+----------+---------------+----------+---------------+
|tagShortID|ListenerShortID|Timestamp |RSSI_Weight_avg|
+----------+---------------+----------+---------------+
|3         |102            |1496745907|80             |
|4         |106            |1496745908|60             |
|2         |101            |1496745909|90             |
+----------+---------------+----------+---------------+

不过，您可以改进上面的代码。希望回答对你有帮助