假设我有一个像这样的数据框。
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 的最后一个非空值。这是我的窗户
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))