apache-spark - 使用 Spark 窗口函数获取最后一个值

Question

假设我有一个这样的数据框。

val df = sc.parallelize(Seq(
            (1.0, 1,"Matt"), 
            (1.0, 2,"John"),
            (1.0, 3,null.asInstanceOf[String]),
            (-1.0, 2,"Adam"), 
            (-1.0, 4,"Steve"))
          ).toDF("id", "timestamp","name")

我想获取按时间戳排序的每个 id 的最后一个非空值。这是我的 window

val partitionWindow = Window.partitionBy($"id").orderBy($"timestamp".desc)

我正在创建一个独特的窗口数据

val filteredDF = df.filter($"name".isNotNull).withColumn("firstName", first("name") over (partitionWindow)).drop("timestamp","name").distinct

并将其加入到实际数据中

val joinedDF = df.join(filteredDF, windowDF.col("id") === filteredDF.col("id")).drop(filteredDF.col("id"))

joinedDF.show()

效果很好，但我不喜欢这个解决方案，谁能给我更好的建议？

另外，谁能告诉我为什么最后一个功能不起作用？我试过了，结果不正确

 val partitionWindow = Window.partitionBy($"id").orderBy($"timestamp")

val windowDF = df.withColumn("lastName", last("name") over (partitionWindow))

Answer 1

如果你想传播最后一个已知值(它与 join 使用的逻辑不同)，你应该:

• 按时间戳排序。
• 采用 last 忽略 nulls:

val partitionWindow = Window.partitionBy($"id").orderBy($"timestamp")

df.withColumn("lastName", last("name", true) over (partitionWindow)).show
// +----+---------+-----+--------+
// |  id|timestamp| name|lastName|
// +----+---------+-----+--------+
// |-1.0|        2| Adam|    Adam|
// |-1.0|        4|Steve|   Steve|
// | 1.0|        1| Matt|    Matt|
// | 1.0|        2| John|    John|
// | 1.0|        3| null|    John|
// +----+---------+-----+--------+

如果要全局取最后一个值:

• 按时间戳排序。
• 设置无界框架。
• 采用 last 忽略 nulls:

val partitionWindow = Window.partitionBy($"id").orderBy($"timestamp")
  .rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)

df.withColumn("lastName", last("name", true) over (partitionWindow)).show
// +----+---------+-----+--------+
// |  id|timestamp| name|lastName|
// +----+---------+-----+--------+
// |-1.0|        2| Adam|   Steve|
// |-1.0|        4|Steve|   Steve|
// | 1.0|        1| Matt|    John|
// | 1.0|        2| John|    John|
// | 1.0|        3| null|    John|
// +----+---------+-----+--------+