Spark Scala按customer分组取timestamp最大值时如何保留所有列

现有修改方案的可行性 #

你后续调整的把所有需要保留的字段都加入聚合Seq用last函数处理的方案是可行的。
实现逻辑是：你先对全量数据按timestamp升序排序，按customer分组后，last函数会取每个分组内排序后的最后一条记录的对应字段值，刚好匹配你要取最新时间对应全字段的需求。
但该方案存在两个注意点：

• 如果对应字段存在空值，last函数默认会跳过空值取最近的非空值，可能出现取到的字段不属于同一条最新记录的问题，建议显式加上ignoreNulls = false参数：

  val val_columns = Seq("value.lastRefresh", "value.associations")
      .map(x => last(col(x), ignoreNulls = false).alias(x))
  

• 低版本Spark（2.3及以下）中groupBy的shuffle操作可能会破坏之前orderBy的排序结果，导致last取到的值不符合预期，低版本环境不建议使用该实现。

更优的实现方案 #

推荐使用窗口函数实现该需求，逻辑更稳定、可读性更高，且不受Spark版本限制：

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.row_number

// 定义窗口规则：按customer分组，按timestamp降序排序
val customerWindow = Window.partitionBy("value.customer").orderBy(col("value.timestamp").desc)

val result = readRdd.select(col("value"))
  .withColumn("row_num", row_number().over(customerWindow))
  .filter(col("row_num") === 1) // 取每个分组排序后的第一条，即时间最新的记录
  .drop("row_num") // 移除辅助计算的行号列

result.show()

该方案的优势：

• 不需要手动罗列所有要保留的字段，自动保留原始数据的所有列
• 逻辑显式可控，不会出现排序被shuffle打乱的问题
• 性能和原聚合方案基本持平，底层都是一次shuffle操作，无额外开销

如果你使用的是Spark 3.0及以上版本，还可以用max_by函数简化聚合写法，性能同样优秀：

val result = readRdd.select(col("value"))
  .groupBy("value.customer")
  .agg(
    max_by(struct("value.timestamp", "value.associations"), col("value.timestamp")).alias("latest")
  )
  .select("latest.*")

内容的提问来源于stack exchange，提问作者Ryan