如何在Spark DataFrame中获得最新价值

Question

我有一个带有以下结构的火花数据框

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

我需要创建一个具有标志1的最新值并在标志0行中更新的数据框。

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

我们有2行flag = 0。考虑第一行（flag = 0），我将有2个值（200和300），我将采用最近的200（2014）。最后一行我没有FLAG 1的最新值，因此使用NULL对其进行了更新。

寻找使用Scala的解决方案。任何帮助将不胜感激。谢谢

Answer 1

您可以尝试使用窗口函数。基本上创建一个窗口，您可以在其中按ID和订单按日期进行分区。然后获取每行的上线。最后，使用何时/否则将所有flag 1的案例转换为null。

类似的东西：

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