PySpark DataFrame去重：保留最新start_date行及所有列

需求说明 #

当employee_name和department同时相同时视为重复行，仅保留start_date最新的一行，且需保留所有列数据。

原始DataFrame #

employee_name   department      start_date                  favorite_color
---------------------------------------------------------------------------
James           sales           2023-11-03T12:38:02.371076Z green
James           sales           2020-01-03T12:38:02.371076Z red
Fred            operations      2023-09-27T16:28:01.920063Z red
Fred            leadership      2023-07-04T16:28:01.920063Z green
Fred            operations      2023-10-27T16:28:01.920063Z green
Wilma           administrative  2021-06-17T16:28:01.920063Z green

尝试的方法及问题 #

使用groupBy+agg(max(start_date))只能获取每组最新日期，但丢失了favorite_color等其他列：

deduped_df = employee_df.groupBy(col("employee_name"), col("department")).agg(max(col("start_date")))

得到的结果：

employee_name   department      start_date                  
-----------------------------------------------------------
Fred            leadership      2023-07-04T16:28:01.920063Z
Fred            operations      2023-10-27T16:28:01.920063Z
James           sales           2023-11-03T12:38:02.371076Z
Wilma           administrative  2021-06-17T16:28:01.920063Z

解决方案 #

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

通过窗口函数按分组字段分区，对start_date降序排序后取每组第一行，完整保留所有列：

from pyspark.sql import Window
from pyspark.sql.functions import row_number, col

# 定义窗口规则：按employee_name和department分组，start_date降序排列
window_spec = Window.partitionBy("employee_name", "department").orderBy(col("start_date").desc())

# 添加行号字段，筛选出每组第一行后删除行号
deduped_df = employee_df.withColumn("row_num", row_number().over(window_spec)) \
                        .filter(col("row_num") == 1) \
                        .drop("row_num")

# 查看结果
deduped_df.show()

最终结果：

employee_name   department      start_date                  favorite_color
---------------------------------------------------------------------------
James           sales           2023-11-03T12:38:02.371076Z green
Fred            leadership      2023-07-04T16:28:01.920063Z green
Fred            operations      2023-10-27T16:28:01.920063Z green
Wilma           administrative  2021-06-17T16:28:01.920063Z green

注意：如果同一分组内有多个行的start_date完全相同（并列最新），row_number()会随机保留其中一行。若需保留所有并列最新行，可改用rank()或dense_rank()代替row_number()。

方法二：关联原表找回字段 #

先通过groupBy获取每组最新日期，再关联原表匹配完整行数据：

# 第一步：获取每组最新的start_date
latest_dates = employee_df.groupBy("employee_name", "department") \
                          .agg(max("start_date").alias("latest_start_date"))

# 第二步：关联原表，筛选出匹配的完整行
deduped_df = employee_df.join(
    latest_dates,
    (employee_df.employee_name == latest_dates.employee_name) &
    (employee_df.department == latest_dates.department) &
    (employee_df.start_date == latest_dates.latest_start_date),
    "inner"
).drop(latest_dates.employee_name, latest_dates.department, latest_dates.latest_start_date)

这种方法会保留所有start_date并列最新的行，适合需要保留全部并列记录的场景。

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