PySpark去重：保留指定列重复行中col2值最大的记录

这里提供两种常用的PySpark实现方案，满足你对['id','col1','col3','col4']列子集去重、保留重复行中col2最大值记录的需求：

方法一：窗口函数（推荐，简洁高效） #

利用窗口函数对指定列分组后排序，取每组的第一条记录：

1. 导入必要模块并初始化SparkSession

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import row_number

spark = SparkSession.builder.appName("DeduplicateMaxCol2").getOrCreate()

2. 创建示例数据集（替换为你的实际数据）

data = [
    (1, 1, 5, 2, 3),
    (1, 1, 0, 2, 3),
    (2, 3, 1, 7, 7),
    (3, 6, 1, 3, 3),
    (3, 6, 5, 3, 3)
]
columns = ["id", "col1", "col2", "col3", "col4"]
df = spark.createDataFrame(data, columns)

3. 定义窗口规范：按目标列分区，按col2降序排序

window_spec = Window.partitionBy("id", "col1", "col3", "col4").orderBy(df["col2"].desc())

4. 添加行号并过滤出每组第一条记录

# 为每个分组的行添加行号，最大值行的行号为1
df_with_row_num = df.withColumn("row_num", row_number().over(window_spec))
# 筛选行号为1的记录，再删除辅助列
result_df = df_with_row_num.filter(df_with_row_num["row_num"] == 1).drop("row_num")

# 查看结果
result_df.show()

方法二：GroupBy + Join #

先分组计算每组的最大col2值，再关联原数据集筛选匹配记录：

from pyspark.sql import SparkSession
from pyspark.sql.functions import max

spark = SparkSession.builder.appName("DeduplicateMaxCol2").getOrCreate()

# 同方法一创建DataFrame
data = [
    (1, 1, 5, 2, 3),
    (1, 1, 0, 2, 3),
    (2, 3, 1, 7, 7),
    (3, 6, 1, 3, 3),
    (3, 6, 5, 3, 3)
]
columns = ["id", "col1", "col2", "col3", "col4"]
df = spark.createDataFrame(data, columns)

# 分组计算每组的最大col2值
max_col2_df = df.groupBy("id", "col1", "col3", "col4").agg(max("col2").alias("max_col2"))

# 关联原数据集，筛选col2等于最大值的记录
result_df = df.join(max_col2_df, on=["id", "col1", "col3", "col4"], how="inner") \
              .filter(df["col2"] == max_col2_df["max_col2"]) \
              .drop("max_col2")

result_df.show()

两种方法的区别 #

• 窗口函数：当同一分组内存在多条col2等于最大值的记录时，只会保留其中一条（由排序规则决定），适合需要唯一结果的场景。
• GroupBy+Join：会保留所有col2等于最大值的记录，适合允许存在多条最大值记录的场景。

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