Spark中能否用DataFrame.dropDuplicates保留最新数据？规则是什么？

一、dropDuplicates('id')的记录保留规则 #

dropDuplicates('id')**不会固定保留最早或最新的记录，最终保留的结果是随机的**。原因在于： Spark是分布式计算框架，数据会被拆分到多个分区中处理。当执行dropDuplicates时，每个分区内遇到重复的id时，只会保留该分区里**第一条出现的记录**，但数据的分区分布、物理存储顺序不受你之前orderBy`的影响（逻辑排序不代表物理存储有序），所以整体来看，最终保留的记录没有固定规律，完全取决于数据的分区情况。这就是你用它无法得到预期结果的核心原因。

二、实现每个id保留最新记录的正确方法 #

方法1：窗口函数法（最直观，适合多字段场景） #

通过窗口函数按id分区、time降序排序，然后取每个分区的第一条记录：

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

// 定义窗口规则：按id分组，time倒序排列
val windowSpec = Window.partitionBy("id").orderBy($"time".desc)

// 给每条记录标记行号，筛选出行号为1的记录（即每个id的最新一条）
val resultDF = df.withColumn("row_num", row_number().over(windowSpec))
  .filter($"row_num" === 1)
  .drop("row_num")

resultDF.show()

方法2：GroupBy+聚合关联法（字段较少时更简洁） #

先分组找到每个id的最大time，再关联回原表获取对应value：

import org.apache.spark.sql.functions.max

// 先获取每个id的最新时间
val maxTimeDF = df.groupBy("id").agg(max("time").alias("max_time"))
// 关联原表筛选出对应记录
val resultDF = maxTimeDF.join(df, Seq("id"), "inner")
  .filter($"time" === $"max_time")
  .drop("max_time")

resultDF.show()

注：如果存在同一个id对应多条相同最大time的记录，以上两种方法会保留所有符合条件的记录，若需进一步去重，可在窗口函数中增加其他排序字段（如value）来确定唯一记录。

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