group by and filter highest value in data frame in scala

user_8235369 0
Viewed 11079

I have some data like this:

a,timestamp,list,rid,sbid,avgvalue
1,1011,1001,4,4,1.20
2,1000,819,2,3,2.40
1,1011,107,1,3,5.40
1,1021,819,1,1,2.10

In the data above I want to find which stamp has the highest tag value (avg. value) based on the tag. Like this.

For time stamp 1011 and a 1:

1,1011,1001,4,4,1.20
1,1011,107,1,3,5.40

The output would be:

1,1011,107,1,3,5.40  //because for timestamp 1011 and tag 1 the higest avg value is 5.40

So I need to pick this column.

I tried this statement, but still it does not work properly:

val highvaluetable = df.registerTempTable("high_value")
val highvalue = sqlContext.sql("select a,timestamp,list,rid,sbid,avgvalue  from high_value")   highvalue.select($"a",$"timestamp",$"list",$"rid",$"sbid",$"avgvalue".cast(IntegerType).as("higher_value")).groupBy("a","timestamp").max("higher_value")
 highvalue.collect.foreach(println)

Any help will be appreciated.

After I applied some of your suggestions, I am still getting duplicates in my data.

+---+----------+----+----+----+----+
|a| timestamp| list|rid|sbid|avgvalue|
+---+----------+----+----+----+----+
|  4|1496745915| 718|   4|   3|0.30|
|  4|1496745918| 362|   4|   3|0.60|
|  4|1496745913| 362|   4|   3|0.60|
|  2|1496745918| 362|   4|   3|0.10|
|  3|1496745912| 718|   4|   3|0.05|
|  2|1496745918| 718|   4|   3|0.30|
|  4|1496745911|1901|   4|   3|0.60|
|  4|1496745912| 718|   4|   3|0.60|
|  2|1496745915| 362|   4|   3|0.30|
|  2|1496745915|1901|   4|   3|0.30|
|  2|1496745910|1901|   4|   3|0.30|
|  3|1496745915| 362|   4|   3|0.10|
|  4|1496745918|3878|   4|   3|0.10|
|  4|1496745915|1901|   4|   3|0.60|
|  4|1496745912| 362|   4|   3|0.60|
|  4|1496745914|1901|   4|   3|0.60|
|  4|1496745912|3878|   4|   3|0.10|
|  4|1496745912| 718|   4|   3|0.30|
|  3|1496745915|3878|   4|   3|0.05|
|  4|1496745914| 362|   4|   3|0.60|
+---+----------+----+----+----+----+

 4|1496745918| 362|   4|   3|0.60|  
 4|1496745918|3878|   4|   3|0.10|

Same time stamp with same tag. This is considered as duplicate.

This is my code:

rdd.createTempView("v1")
val rdd2=sqlContext.sql("select max(avgvalue) as max from v1 group by  (a,timestamp)")
rdd2.createTempView("v2")
val rdd3=sqlContext.sql("select a,timestamp,list,rid,sbid,avgvalue from v1  join v2 on v2.max=v1.avgvalue").show()
apache-spark-sql apache-spark scala sql
4 Answers

All the other solutions provided here did not give me the correct answer so this is what it worked for me with row_number():

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window

val windowSpec = Window.partitionBy("timestamp").orderBy(desc("avgvalue"))

df.select("a", "timestamp", "list", "rid", "sbid", "avgvalue")
  .withColumn("largest_avgvalue", row_number().over( windowSpec ))
  .filter($"largest_avgvalue" === 1)
  .drop("largest_avgvalue")

The other solutions had the following problems in my tests:

  • The solution with .agg( max(x).as(x), first(y).as(y), ... ) doesn't work because first() function "will return the first value it sees" according to documentation, which means it is non-deterministic,
  • The solution with .withColumn("x", max("y") over windowSpec.orderBy("m") ) doesn't work because the result of the max will be same as in the value that is selecting for the row. I believe the problem there is the orderBy()".

Hence, the following also gives the correct answer, with max():

val windowSpec = Window.partitionBy("timestamp").orderBy(desc("avgvalue"))
    
df.select("a", "timestamp", "list", "rid", "sbid", "avgvalue")
  .withColumn("largest_avgvalue", max("avgvalue").over( windowSpec ))
  .filter($"largest_avgvalue" === $"avgvalue")
  .drop("largest_avgvalue")
user_5501550
Related