Datenrahmen Datensätze Abfrage Ausgabe

Question

Env: Spark-1.6, Scala

Hallo
Ich habe Aufzeichnungen in Datenrahmen wie unten

reportTime    serverNAme  channel  viewer 
2017-01-12 19:16:58.76 MTRL28DSERV722 Channel1 1192 
2017-01-12 19:16:59.213 MTRL28DSERV722 Channel1 668 
2017-01-12 19:17:05.193 BBBN23DSERV000 Channel1 795 
2017-01-12 19:17:01.15 BBBN23DSERV000 Channel1 700 

2017-01-12 19:16:58.76 MTRL28DSERV722 Channel3 100 
2017-01-12 19:16:59.213 MTRL28DSERV722 Channel3 110 
2017-01-12 19:17:05.193 BBBN23DSERV000 Channel3 200 
2017-01-12 19:17:01.15 BBBN23DSERV000 Channel3 50 

Ich brauche Ausgabe: neueste Zuschauer von jedem Server von Channel-, so dass

2017-01-12 19:16:59.213 MTRL28DSERV722 Channel1 668 
2017-01-12 19:17:05.193 BBBN23DSERV000 Channel1 795 


2017-01-12 19:16:59.213 MTRL28DSERV722 Channel3 100 
2017-01-12 19:17:05.193 BBBN23DSERV000 Channel3 200 

Wie kann ich ein Ergebnis erhalten? Ich möchte DF nicht als verführbar registrieren und SQL dafür verwenden, da SQL in Spark 1.6 nicht optimiert ist.
Dank
Hossain

Answer 1

Sie ein row_number innerhalb jeder Kombination von severName und channel, in absteigender Reihenfolge der reportTime erstellen können, und dann where verwenden Sie die erste Zeile in jeder Gruppe zu filtern, die den mit den jüngsten sein sollten reportTime:

import org.apache.spark.sql.expressions.Window 
import org.apache.spark.sql.functions.row_number  
val w = Window.partitionBy($"serverNAme", $"channel").orderBy($"reportTime".desc) 

df.withColumn("rn", row_number.over(w)).where($"rn" === 1).drop($"rn").show 

+--------------------+--------------+--------+------+ 
|   reportTime| serverNAme| channel|viewer| 
+--------------------+--------------+--------+------+ 
|2017-01-12 19:17:...|BBBN23DSERV000|Channel3| 200| 
|2017-01-12 19:16:...|MTRL28DSERV722|Channel3| 110| 
|2017-01-12 19:17:...|BBBN23DSERV000|Channel1| 795| 
|2017-01-12 19:16:...|MTRL28DSERV722|Channel1| 668| 

+--------------------+--------------+--------+------+