PySpark中高效移除重复数据的两种策略

DataFrame.dropDuplicates(subset=None)

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

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

# 创建一个示例PySpark SQL DataFrame
data = [("C001", "Alice"), ("C002", "Bob"), ("C001", "Alice"), ("C003", "Charlie"), ("C002", "Bob")]
columns = ["CUSTOMER_ID", "NAME"]
df_sql = spark.createDataFrame(data, columns)

print("原始 PySpark SQL DataFrame:")
df_sql.show()

# 1. 对所有列进行去重
df_distinct_all = df_sql.dropDuplicates()
print("所有列去重后的 DataFrame:")
df_distinct_all.show()

# 2. 仅根据 'CUSTOMER_ID' 列进行去重
# 注意：当仅根据子集去重时，对于重复的子集行，Spark会保留其中任意一行，其非子集列的值可能不确定。
# 在此示例中，由于(C001, Alice)是完全重复的，所以行为一致。
# 但如果数据是 (C001, Alice) 和 (C001, David)，则去重后会保留其中一个。
df_distinct_id = df_sql.dropDuplicates(subset=["CUSTOMER_ID"])
print("根据 'CUSTOMER_ID' 列去重后的 DataFrame:")
df_distinct_id.show()

# 停止SparkSession
spark.stop()

原始 PySpark SQL DataFrame:
+-----------+-------+
|CUSTOMER_ID|   NAME|
+-----------+-------+
|       C001|  Alice|
|       C002|    Bob|
|       C001|  Alice|
|       C003|Charlie|
|       C002|    Bob|
+-----------+-------+

所有列去重后的 DataFrame:
+-----------+-------+
|CUSTOMER_ID|   NAME|
+-----------+-------+
|       C001|  Alice|
|       C002|    Bob|
|       C003|Charlie|
+-----------+-------+

根据 'CUSTOMER_ID' 列去重后的 DataFrame:
+-----------+-------+
|CUSTOMER_ID|   NAME|
+-----------+-------+
|       C001|  Alice|
|       C002|    Bob|
|       C003|Charlie|
+-----------+-------+

DataFrame.drop_duplicates(subset=None, keep='first', inplace=False, ignore_index=False)

import pyspark.pandas as ps
from pyspark.sql import SparkSession

# 初始化SparkSession (pyspark.pandas 会自动使用现有的SparkSession)
spark = SparkSession.builder.appName("DropDuplicatesPandas").getOrCreate()

# 创建一个示例PySpark Pandas DataFrame
data = {"CUSTOMER_ID": ["C001", "C002", "C001", "C003", "C002"],
        "NAME": ["Alice", "Bob", "Alice", "Charlie", "Bob"]}
psdf = ps.DataFrame(data)

print("原始 PySpark Pandas DataFrame:")
print(psdf)

# 1. 对所有列进行去重 (默认 keep='first')
psdf_distinct_all = psdf.drop_duplicates()
print("所有列去重后的 DataFrame:")
print(psdf_distinct_all)

# 2. 仅根据 'CUSTOMER_ID' 列进行去重，保留第一个
psdf_distinct_id_first = psdf.drop_duplicates(subset=["CUSTOMER_ID"], keep='first')
print("根据 'CUSTOMER_ID' 列去重 (保留第一个) 后的 DataFrame:")
print(psdf_distinct_id_first)

# 3. 仅根据 'CUSTOMER_ID' 列进行去重，保留最后一个
psdf_distinct_id_last = psdf.drop_duplicates(subset=["CUSTOMER_ID"], keep='last')
print("根据 'CUSTOMER_ID' 列去重 (保留最后一个) 后的 DataFrame:")
print(psdf_distinct_id_last)

# 4. 仅根据 'CUSTOMER_ID' 列进行去重，删除所有重复项
psdf_distinct_id_false = psdf.drop_duplicates(subset=["CUSTOMER_ID"], keep=False)
print("根据 'CUSTOMER_ID' 列去重 (删除所有重复项) 后的 DataFrame:")
print(psdf_distinct_id_false)

# 停止SparkSession (如果需要，但通常在脚本结束时自动停止)
spark.stop()

原始 PySpark Pandas DataFrame:
  CUSTOMER_ID     NAME
0        C001    Alice
1        C002      Bob
2        C001    Alice
3        C003  Charlie
4        C002      Bob

所有列去重后的 DataFrame:
  CUSTOMER_ID     NAME
0        C001    Alice
1        C002      Bob
3        C003  Charlie

根据 'CUSTOMER_ID' 列去重 (保留第一个) 后的 DataFrame:
  CUSTOMER_ID     NAME
0        C001    Alice
1        C002      Bob
3        C003  Charlie

根据 'CUSTOMER_ID' 列去重 (保留最后一个) 后的 DataFrame:
  CUSTOMER_ID     NAME
2        C001    Alice
4        C002      Bob
3        C003  Charlie

根据 'CUSTOMER_ID' 列去重 (删除所有重复项) 后的 DataFrame:
  CUSTOMER_ID     NAME
3        C003  Charlie