在日常的数据处理工作中，我们经常会被杂乱无章的数据困扰。作为一名大数据工程师，我发现在实际的数据集中，往往包含大量重复的行或数据点。这些重复值不仅会占用宝贵的存储空间，更有可能导致我们的分析结果产生偏差，甚至在机器学习模型训练中造成数据泄露的问题。因此，清洗数据、去除冗余是数据预处理流程中至关重要的一步。
在本文中，我们将深入探讨 PySpark 中处理重复数据的核心策略——即如何保留数据的第一次出现，同时精准地删除其余的重复项。我们将通过实战案例，从基础概念到性能优化，并结合 2026 年最新的工程化理念，全面掌握 dropDuplicates() 函数的用法。这不仅是一次语法学习，更是一次关于如何编写高质量、高效率数据清洗代码的探讨。
为什么去重如此重要？
在开始编写代码之前，我想强调一下为什么我们需要如此重视去重操作。想象一下，如果你正在分析用户的购买行为，而由于微服务间的网络抖动，同一笔订单被记录了两次。如果你不去重，计算出的总销售额将是虚高的，直接影响到财务报表的准确性。PySpark 作为分布式计算引擎，处理 PB 级数据时，一个微小的逻辑错误可能导致巨大的资源浪费。因此，掌握正确的去重方法，是我们保证数据准确性的第一道防线。
认识 dropDuplicates() 函数
PySpark 的 DataFrame API 为我们提供了一个非常直观且强大的函数：dropDuplicates()。这个函数的设计初衷非常简单：根据指定的列（或所有列）判断数据是否重复，并默认保留每组重复数据中的第一条记录。
核心语法：
dataframe_name.dropDuplicates(column_names)
这里的 INLINECODEc09ae219 是一个可选参数。如果不传入参数，PySpark 会检查所有列的内容完全一致时才判定为重复；如果传入了列名列表（例如 INLINECODE43ffbd11），则只有当这些指定列的值同时相同时，才会被视为重复数据。这种灵活性让我们能够应对各种复杂的业务逻辑。
环境准备与数据构建
为了演示如何去重，我们首先需要构建一个包含脏数据的演示环境。让我们定义一个包含学生信息的数据集，其中特意加入了一些显而易见的重复项，以便我们观察处理前后的变化。
我们将创建一个包含姓名、学号、班级ID、分数和课外活动的 DataFrame。
# 导入必要的 PySpark 模块
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
# 构建 Spark 会话（这是所有操作的起点）
# 在 2026 年的本地开发环境中，我们通常会在连接云端集群前进行本地模拟
spark = SparkSession.builder \
.appName("DataDeduplicationDemo") \
.master("local[2]") # 本地模式，使用2个核心
.config("spark.sql.adaptive.enabled", "true") # 启用自适应查询执行
.getOrCreate()
# 模拟原始数据（注意观察 Ritika 和 Atirikt 的数据）
raw_data = [
("Pulkit", 12, "CS32", 82, "Programming"),
("Ritika", 20, "CS32", 94, "Writing"), # Ritika 的第一次出现
("Ritika", 20, "CS32", 84, "Writing"), # Ritika 的重复（分数不同）
("Atirikt", 4, "BB21", 58, "Doctor"), # Atirikt 的第一次出现
("Atirikt", 4, "BB21", 78, "Doctor"), # Atirikt 的重复
("Ghanshyam", 4, "DD11", 38, "Lawyer"), # 学号与其他人重复，但姓名唯一
("Reshav", 18, "EE43", 56, "Timepass")
]
# 定义 Schema 以确保数据类型清晰
# 最佳实践：总是显式定义 Schema，避免 Spark 推断带来的开销
schema = StructType([
StructField("Name", StringType(), True),
StructField("Roll Number", IntegerType(), True),
StructField("Class ID", StringType(), True),
StructField("Marks", IntegerType(), True),
StructField("Extracurricular", StringType(), True)
])
# 创建 DataFrame
student_df = spark.createDataFrame(data=raw_data, schema=schema)
# 展示原始数据，直观感受“脏数据”的样子
print("--- 原始数据视图 ---")
student_df.show(truncate=False)
输出结果预览：
在运行上述代码后，你将看到名为 Ritika 和 Atirikt 的学生各出现了两次。虽然他们的基本信息一致，但 Marks（分数）字段存在差异。这引出了一个关键问题：当存在部分字段冲突时，我们保留哪一行？默认情况下，PySpark 会保留最先出现的那个。
示例 1：基于单列去重
这是最常见的场景：我们需要根据特定的唯一标识符（例如学号 Roll Number）来去除重复。也许在数据录入时，同一个学号被误录了多次，但我们只关心唯一的学生记录。
让我们来看看如何只基于“Roll Number”这一列进行去重。
# 仅基于 ‘Roll Number‘ 列进行去重
# 逻辑：如果学号相同，则视为重复，保留第一条。
df_unique_roll = student_df.dropDuplicates([‘Roll Number‘])
print("--- 基于 ‘Roll Number‘ 去重后的结果 ---")
df_unique_roll.show(truncate=False)
代码解析：
请注意观察输出结果。我们会发现 Ritika (20) 和 Atirikt (4) 的重复记录消失了。但是，保留的是哪一行呢？
- 对于 Ritika，保留了 Marks 为 94 的记录（第一次出现）。
- 对于 Atirikt，保留了 Marks 为 58 的记录。
这就引出了一个实际开发中的陷阱：去重是“无序”的保留策略。在分布式计算中，数据的“第一次出现”取决于数据分区的顺序。如果你需要保留特定条件下的记录（比如保留分数更高的那一行），直接使用 dropDuplicates 是不够的，我们后面会讨论解决方案。
示例 2：基于多列组合去重
现实业务逻辑往往比单层去重要复杂得多。比如，我们可能认为只有当“姓名”和“学号”都完全一样时，才算作真正的重复（毕竟可能存在同名同姓但学号不同的学生，或者学号不同但名字相同的情况）。
让我们尝试同时基于 [‘Roll Number‘, ‘Name‘] 进行去重。
# 基于 ‘Roll Number‘ 和 ‘Name‘ 两列组合去重
# 逻辑：只有当学号和姓名都相同时，才判定为重复。
df_unique_combo = student_df.dropDuplicates([‘Roll Number‘, ‘Name‘])
print("--- 基于 ‘Roll Number‘ 和 ‘Name‘ 去重后的结果 ---")
df_unique_combo.show(truncate=False)
深度解析：
仔细查看输出，你会发现 Ghanshyam 这一行被保留了。虽然他的 Roll Number (4) 与 Atirikt 重复，但因为我们在函数中指定了两个参数，PySpark 会同时检查这两个列。由于 Name 不同，Ghanshyam 被认为是唯一的数据点。
这展示了 dropDuplicates 的组合判断能力：只有当所有传入参数的值都发生重复时，删除操作才会触发。
示例 3：全局去重
有时候，我们需要一种更为激进的策略：完全相同的行。也就是说，只有当行中所有字段（从 Name 到 Extracurricular）的值都一模一样时，才删除。
要实现这一点，我们只需要不传任何参数，或者传入 [] 即可。
# 全局去重：检查所有列
# 如果不传参数，PySpark 会比较所有字段
df_all_cols = student_df.dropDuplicates()
print("--- 全局完全去重后的结果 ---")
df_all_cols.show(truncate=False)
实战见解：
在这个特定的数据集中，你会发现即使 Ritika 的两行数据看起来很“重复”，但由于 Marks（94 和 84）不同，它们在全局去重下都被保留了。这是非常有用的特性，它防止了我们误删那些看似重复但实则包含关键差异信息的数据。
进阶技巧：如何删除除最后一条之外的所有记录？
很多开发者会问：“我习惯于保留最新的记录，而不是最早的，该怎么办？”
PySpark 的 INLINECODE722f91dd 默认保留第一条，如果我们想保留最后一条，通常的做法是使用“窗口函数”配合 INLINECODE59277768。虽然这不是 dropDuplicates 的直接功能，但作为进阶知识，我强烈建议你掌握这种方法，因为它提供了更精细的控制力。
from pyspark.sql.window import Window
from pyspark.sql.functions import col, row_number
# 1. 定义窗口规范：按 ‘Roll Number‘ 分区，按时间或ID排序（这里假设 Marks 大的代表新数据）
# 在生产环境中，这里应该是一个明确的时间戳字段
window_spec = Window.partitionBy("Roll Number").orderBy(col("Marks").desc())
# 2. 添加行号
df_with_rank = student_df.withColumn("rank", row_number().over(window_spec))
# 3. 过滤出 rank = 1 的记录（即每组中 Marks 最高的那一行）
df_keep_last = df_with_rank.filter(col("rank") == 1).drop("rank")
print("--- 保留每组中 Marks 最高的记录 ---")
df_keep_last.show(truncate=False)
通过这种方式，我们绕过了“保留第一条”的限制，实现了保留“特定条件下的一条”的高级需求。
深入解析：2026年视角下的去重性能优化
在 2026 年，数据规模呈指数级增长，简单的 API 调用可能无法满足性能需求。让我们深入探讨 dropDuplicates 背后的机制，并分享我们在生产环境中的优化策略。
#### 1. Shuffle 与 数据倾斜 的博弈
dropDuplicates() 本质上是一个宽依赖操作。Spark 必须根据指定的列计算哈希值，将相同键的数据通过网络传输到同一个 Executor 上。这涉及到巨大的 Shuffle 开销。
工程化建议：
在生产环境中，如果发现去重任务极其缓慢，请检查是否发生了数据倾斜。如果某个 Key（例如某个特定的 Roll Number）的数据量远超其他 Key，会导致处理该 Key 的 Executor 成为瓶颈。
解决方案：加盐
我们可以在去重前引入随机前缀来打散数据，处理后再聚合。虽然增加了步骤，但在处理极度倾斜的数据时，往往能将耗时从数小时降低到几分钟。
from pyspark.sql.functions import rand, floor, concat, lit
# 假设 Roll Number = 1 是个热点数据，有 1000 万条重复
# 我们添加一个随机前缀 (0-9)
salt_count = 10
df_salt = student_df.withColumn("salt", (floor(rand() * salt_count)).cast("string"))
df_salt = df_salt.withColumn("pseudo_key", concat(col("salt"), lit("_"), col("Roll Number")))
# 针对伪 Key 进行去重（这步会分散压力）
df_dedup_salt = df_salt.dropDuplicates(["pseudo_key"])
# 最后再次对原始 Key 进行全局去重，合并各分片结果
# 注意：这里的逻辑需要根据具体业务调整，仅作原理演示
# 最终步骤通常还需要再次聚合以确保全局唯一性
#### 2. 物化视图与增量去重
如果您的数据是源源不断的流式数据（例如点击流日志），每次都对全量历史数据运行 dropDuplicates 是极其低效的。
现代架构方案：
我们建议构建“只增”的 Delta Lake 或 Hudi 表。利用这些现代数据湖技术的 INLINECODEb73627ff 和 INLINECODE585356cd 功能，可以在写入时自动处理去重，而无需每次都运行显式的去重作业。
AI 辅助与开发体验（2026 趋势）
在这一节，让我们聊聊如何利用当下的 AI 工具来提升编写 PySpark 代码的效率。
#### 利用 Cursor / Copilot 生成复杂逻辑
当我们需要编写复杂的窗口函数去重逻辑时，与其查阅晦涩的文档，不如直接向 AI 描述意图。
Prompt 示例：
> “我有一个 Spark DataFrame，包含 INLINECODE0decdf8e 和 INLINECODE681aed9f。请写一段 PySpark 代码，为每个用户保留最新的两条记录（保留最新和次新），删除其他所有记录。请使用 Window 函数实现。”
通过这种 Vibe Coding（氛围编程） 方式，AI 工具（如 Cursor）不仅会生成代码，还会帮你处理导入语句和注释。我们团队现在经常让 AI 作为一个“耐心的副驾驶”，帮我们编写单元测试来验证去重逻辑的边界情况（例如：输入为空 DataFrame 时会发生什么？）。
性能基准测试与监控
最后，作为资深工程师，我们必须“相信数据，而不是直觉”。让我们通过代码来对比不同方式的性能。
import time
# 创建一个较大的数据集进行压力测试
large_data = [(i, f"Name_{i % 1000}", f"Class_{i % 100}") for i in range(1, 1000000)]
large_df = spark.createDataFrame(large_data, ["id", "name", "class"])
large_df.cache() # 缓存以进行公平测试
start_time = time.time()
# 测试 1: 全局去重 (检查所有列)
res1 = large_df.dropDuplicates()
res1.count() # 触发 Action
t1 = time.time() - start_time
start_time = time.time()
# 测试 2: 仅 ID 去重
res2 = large_df.dropDuplicates(["id"])
res2.count()
t2 = time.time() - start_time
print(f"全局去重耗时: {t1:.2f}s, ID去重耗时: {t2:.2f}s")
总结
在本文中，我们不仅学习了如何使用 PySpark 的 dropDuplicates() 函数来清洗数据，更深入探讨了单列去重、多列组合去重以及全局去重的区别。我们通过构建实际的 DataFrame，亲眼见证了数据从杂乱到整洁的过程。
我们还进一步讨论了如何通过窗口函数来实现更复杂的“保留特定记录”的逻辑，这是从初级迈向高级的重要一步。结合 2026 年的技术视角，我们探讨了数据倾斜优化、AI 辅助编码以及性能基准测试的重要性。希望这些知识能帮助你在未来的数据处理项目中，写出更加健壮、高效的代码。
最后，别忘了在运行完所有操作后关闭 Spark 会话，释放资源：
spark.stop()