Wie letzten Wert in Funkendatenrahmen

Question

bekommt ich einen Funken Datenrahmen habe mit Struktur folgenden

id flag price date 
    a 0 100 2015 
    a 0 50 2015 
    a 1 200 2014 
    a 1 300 2013 
    a 0 400 2012 

Ich brauche einen Datenrahmen mit dem jüngsten Wert von Flag 1 und in der Flagge 0 Zeilen aktualisiert zu erstellen.

 id flag price date new_column 
     a 0 100 2015 200 
     a 0 50 2015 200 
     a 1 200 2014 null 
     a 1 300 2013 null 
     a 0 400 2012 null 

Wir haben 2 Zeilen mit Flag = 0. Betrachten Sie die erste Zeile (Flag = 0), ich werde 2 Werte haben (200 und 300) und ich nehme den letzten 200 (2014). Und in der letzten Zeile habe ich keinen aktuellen Wert für Flag 1, daher wird es mit Null aktualisiert.

Auf der Suche nach einer Lösung mit scala. Irgendwelche Hilfe würde geschätzt werden. Dank

Answer 1

Sie können versuchen, Fensterfunktionen zu verwenden. Im Grunde erstellen Sie ein Fenster, in dem Sie nach ID und Reihenfolge nach Datum partitionieren. Holen Sie dann die vorherige Zeile für jede Zeile. Zuletzt verwenden Sie wenn/sonst, um alle Fälle mit Flag 1 auf Null zu setzen.

Etwas wie folgt aus:

val df = sc.parallelize(List(("a",0,100,2015),("a",1,200,2014),("a",1,300,2013),("a",0,400,2012))).toDF("id","flag","price","date") 
import org.apache.spark.sql.expressions.Window 
import org.apache.spark.sql.functions.{lag,when} 
val wSpec1 = Window.partitionBy("id").orderBy("date") 
val df2=df.withColumn("last1",when(df("flag")===0,lag('price, 1).over(wSpec1)).otherwise(null))