问题场景描述
在数据清洗和预处理过程中，我们经常会遇到包含重复记录的数据集。这些重复记录可能基于一个或多个列的组合，但我们往往需要为每个重复组保留特定数量的记录，例如，只保留每个重复组中最新的n条记录。例如，在一个包含用户活动记录的dataframe中，我们可能希望针对每个用户（由first_name, last_name, sex等列定义），只保留其最新的3条活动记录。
假设我们有如下一个DataFrame：
| id | first_name | last_name | sex | country |
|---|---|---|---|---|
| 01 | John | Doe | Male | USA |
| 02 | John | Doe | Male | Canada |
| 03 | John | Doe | Male | Mexico |
| 04 | Mark | Kay | Male | Italy |
| 05 | John | Doe | Male | Spain |
| 06 | Mark | Kay | Male | France |
| 07 | John | Doe | Male | Peru |
| 08 | Mark | Kay | Male | India |
| 09 | Mark | Kay | Male | Laos |
| 10 | John | Doe | Male | Benin |
目标是基于first_name、last_name和sex列的组合识别重复项，并为每个组合保留最新的3条记录（根据id列的降序）。
基于Pandas的解决方案：使用groupby().tail()
对于在内存中操作的Pandas DataFrame，groupby().tail()方法提供了一种非常简洁且高效的解决方案。
核心思路
- 排序数据： 首先，确保DataFrame按照定义“最新”的标准进行排序。在我们的例子中，id越大表示越新，因此需要按id升序排序。
- 分组： 按照用于识别重复项的列（例如first_name, last_name, sex）进行分组。
- 选取尾部记录： 对每个分组应用tail(n)方法，这将返回该分组中最后N条记录。由于我们已经提前排序，这些“最后N条”就是我们想要的“最新N条”。
示例代码
import pandas as pd
# 原始DataFrame数据
data = {
'id': [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
'first_name': ['John', 'John', 'John', 'Mark', 'John', 'Mark', 'John', 'Mark', 'Mark', 'John'],
'last_name': ['Doe', 'Doe', 'Doe', 'Kay', 'Doe', 'Kay', 'Doe', 'Kay', 'Kay', 'Doe'],
'sex': ['Male', 'Male', 'Male', 'Male', 'Male', 'Male', 'Male', 'Male', 'Male', 'Male'],
'country': ['USA', 'Canada', 'Mexico', 'Italy', 'Spain', 'France', 'Peru', 'India', 'Laos', 'Benin']
}
df = pd.DataFrame(data)
# 1. 根据'id'列对DataFrame进行排序，确保'tail(3)'能获取到最新的3条记录
# 如果'id'本身就是递增的，此步骤可确保正确性。
df_sorted = df.sort_values(by='id').copy()
# 2. 按照指定列进行分组，并为每个组保留最后3条记录
result_df = df_sorted.groupby(['first_name', 'last_name', 'sex']).tail(3)
# 3. （可选）重置索引，使索引连续
result_df = result_df.reset_index(drop=True)
# 显示结果DataFrame
print("处理后的DataFrame:")
print(result_df)输出结果：
处理后的DataFrame: id first_name last_name sex country 0 5 John Doe Male Spain 1 6 Mark Kay Male France 2 7 John Doe Male Peru 3 8 Mark Kay Male India 4 9 Mark Kay Male Laos 5 10 John Doe Male Benin
基于PySpark的解决方案：使用窗口函数
对于大规模分布式数据集，例如在Apache Spark环境中使用PySpark，groupby().tail()方法不再适用。此时，窗口函数（Window Functions）是实现此功能的标准且高效的方式。
核心思路
- 定义窗口规范： 使用Window.partitionBy()定义分组的列，并使用orderBy()定义组内排序的列。为了获取“最新”的记录，排序通常是降序（例如，id降序）。
- 生成行号： 在每个分区（即每个重复组）内，根据排序规则为每条记录生成一个行号（row_number()）。
- 过滤： 筛选出row_number小于或等于N的记录。
示例代码（PySpark风格）
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window
# 假设df是一个Spark DataFrame
# 这里为了示例，我们创建一个模拟的SparkSession和DataFrame
spark = SparkSession.builder.appName("FilterDuplicates").getOrCreate()
data = [
(1, 'John', 'Doe', 'Male', 'USA'),
(2, 'John', 'Doe', 'Male', 'Canada'),
(3, 'John', 'Doe', 'Male', 'Mexico'),
(4, 'Mark', 'Kay', 'Male', 'Italy'),
(5, 'John', 'Doe', 'Male', 'Spain'),
(6, 'Mark', 'Kay', 'Male', 'France'),
(7, 'John', 'Doe', 'Male', 'Peru'),
(8, 'Mark', 'Kay', 'Male', 'India'),
(9, 'Mark', 'Kay', 'Male', 'Laos'),
(10, 'John', 'Doe', 'Male', 'Benin')
]
columns = ['id', 'first_name', 'last_name', 'sex', 'country']
df_spark = spark.createDataFrame(data, columns)
# 定义窗口规范：按first_name, last_name, sex分组，按id降序排序
window_spec = Window.partitionBy('first_name', 'last_name', 'sex').orderBy(F.desc('id'))
# 为每个分区内的记录生成行号
df_with_row_number = df_spark.withColumn('row_number', F.row_number().over(window_spec))
# 过滤，只保留行号小于等于3的记录
filtered_df_spark = df_with_row_number.filter('row_number <= 3')
# 移除辅助列row_number
result_df_spark = filtered_df_spark.drop('row_number')
# 显示结果
print("处理后的Spark DataFrame:")
result_df_spark.show()
spark.stop()输出结果：
处理后的Spark DataFrame: +---+----------+---------+----+-------+ | id|first_name|last_name| sex|country| +---+----------+---------+----+-------+ | 5| John| Doe|Male| Spain| | 7| John| Doe|Male| Peru| | 10| John| Doe|Male| Benin| | 6| Mark| Kay|Male| France| | 8| Kay | Kay|Male| India| | 9| Mark| Kay|Male| Laos| +---+----------+---------+----+-------+
效率与选择考量
- Pandas groupby().tail()： 对于数据集能够完全载入内存的情况，groupby().tail()方法通常非常高效且代码简洁。它是Pandas中处理此类问题的首选方法。其内部实现经过高度优化，能够有效处理分组和选择操作。
- PySpark 窗口函数： 对于大规模分布式数据集，当数据量超出单机内存限制时，PySpark的窗口函数是唯一的选择。虽然代码可能比Pandas版本稍长，但它能在分布式集群上高效执行，避免了数据收集到单个节点的瓶颈。Spark的优化器能够智能地处理窗口操作，确保性能。
注意事项
- 排序的重要性： 无论是Pandas还是PySpark，定义“最新”或“最旧”的关键在于正确的排序。如果“最新”是基于时间戳，则应按时间戳列排序。如果“最新”是基于某个ID，则按ID排序。
-
性能优化：
- 在Pandas中，如果DataFrame非常大，sort_values()可能会消耗较多内存和时间。确保你的系统有足够的资源。
- 在PySpark中，窗口操作会涉及数据重分区（shuffle），这可能是一个耗时操作。合理选择partitionBy的列，避免创建过多或过少的分区，有助于优化性能。
- reset_index()： 在Pandas中，groupby().tail()操作会保留原始索引。如果需要一个从0开始的连续索引，记得调用reset_index(drop=True)。
总结
本文详细介绍了在数据处理中，如何根据特定分组筛选重复记录并保留指定数量（N）的最新数据。对于内存中的数据集，Pandas的df.sort_values().groupby().tail(N)组合方法提供了一个简洁高效的解决方案。而对于分布式大数据集，PySpark的窗口函数（Window.partitionBy().orderBy().row_number()）则是实现相同逻辑的标准且高性能途径。理解这两种方法的适用场景和实现原理，能帮助开发者根据实际需求选择最合适的工具和策略，从而高效地完成数据清洗任务。
