Pandas数据处理：高效筛选重复记录并保留指定数量的最新数据

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

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|
+---+----------+---------+----+-------+