PySpark DataFrame去重并保留每组最后一条记录的实现

方法1：使用窗口函数（推荐） #

通过窗口函数按分组键分区、ID降序排序，直接取每组的第一条记录（对应原数据的最后一条），同时自动处理重复项。

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import col, row_number, split

# 初始化SparkSession
spark = SparkSession.builder.appName("KeepLastRecord").getOrCreate()

# 构造输入DataFrame
data = [
    (1, "CAR1_V1"),
    (2, "CAR1_V2"),
    (3, "CAR1_V3"),
    (4, "CAR2_V1"),
    (5, "CAR2_V2"),
    (6, "CAR3_V1"),
    (7, "CAR3_V1")
]
df = spark.createDataFrame(data, ["ID", "Value"])

# 定义窗口规则：按Value下划线前缀分组，按ID降序排序
window_spec = Window.partitionBy(split(col("Value"), "_").getItem(0)).orderBy(col("ID").desc())

# 标记每组行号，筛选行号为1的记录（即每组最后一条）
result_df = df.withColumn("row_num", row_number().over(window_spec)) \
              .filter(col("row_num") == 1) \
              .drop("row_num") \
              .orderBy("ID")

# 查看结果
result_df.show()

方法2：分组取最大ID再过滤 #

先分组计算每组的最大ID（对应最后一条记录的ID），再用这些ID筛选原DataFrame。

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, split, max

# 初始化SparkSession（已初始化可跳过）
spark = SparkSession.builder.appName("KeepLastRecord").getOrCreate()

# 构造输入DataFrame（同上）
data = [
    (1, "CAR1_V1"),
    (2, "CAR1_V2"),
    (3, "CAR1_V3"),
    (4, "CAR2_V1"),
    (5, "CAR2_V2"),
    (6, "CAR3_V1"),
    (7, "CAR3_V1")
]
df = spark.createDataFrame(data, ["ID", "Value"])

# 提取分组键，分组获取每组最大ID
max_id_df = df.withColumn("group_key", split(col("Value"), "_").getItem(0)) \
              .groupBy("group_key") \
              .agg(max("ID").alias("max_id"))

# 关联原DataFrame，筛选出最大ID对应的记录
result_df = df.join(max_id_df, df.ID == max_id_df.max_id, "inner") \
              .select(df.ID, df.Value) \
              .orderBy("ID")

# 查看结果
result_df.show()

输出结果 #

两种方法均会得到目标DataFrame：

+---+-------+
| ID|  Value|
+---+-------+
|  3|CAR1_V3|
|  5|CAR2_V2|
|  7|CAR3_V1|
+---+-------+

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