PySpark 3.5 数据清洗实战：5 个 filter 与 distinct 组合场景解决重复与脏数据
·
PySpark 3.5 数据清洗实战：5 个 filter 与 distinct 组合场景解决重复与脏数据
数据质量是数据分析的生命线。在实际业务场景中，我们常常会遇到各种数据质量问题：重复记录、异常值、缺失值、格式不一致等。这些问题如果不及时处理，会直接影响后续分析的准确性和可靠性。PySpark 作为大数据处理的首选工具之一，提供了丰富的算子来处理这些问题，其中 filter 和 distinct 是最基础但也是最强大的两个算子。
本文将介绍 5 个真实业务场景下 filter 和 distinct 的组合应用，帮助数据工程师和分析师解决常见的数据质量问题。每个场景都配有完整的代码示例和详细的解释，可以直接应用到实际项目中。
1. 场景一：去重保留最新记录
在用户行为分析中，我们经常会遇到同一个用户在同一时间点产生多条记录的情况。这种情况下，我们需要保留最新的记录，同时去除其他重复记录。
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, max as max_
from pyspark.sql.window import Window
# 创建SparkSession
spark = SparkSession.builder.appName("DataCleaning").getOrCreate()
# 模拟用户行为数据
data = [
("user1", "2023-01-01 10:00:00", "click"),
("user1", "2023-01-01 10:00:05", "view"),
("user1", "2023-01-01 10:00:03", "purchase"),
("user2", "2023-01-01 11:00:00", "click"),
("user2", "2023-01-01 11:00:01", "click")
]
df = spark.createDataFrame(data, ["user_id", "timestamp", "action"])
# 定义窗口函数，按用户分组并按时间戳降序排序
window = Window.partitionBy("user_id").orderBy(col("timestamp").desc())
# 添加行号列，然后过滤出行号为1的记录（即每个用户最新的记录）
result = df.withColumn("row_num", row_number().over(window)) \
.filter(col("row_num") == 1) \
.drop("row_num")
result.show()
关键点解析：
- 使用
Window.partitionBy按用户ID分组 - 使用
orderBy按时间戳降序排序 - 使用
row_number()为每条记录添加行号 - 使用
filter保留行号为1的记录（最新记录）
2. 场景二：过滤异常值后统计唯一用户
在分析用户活跃度时，我们需要先过滤掉异常值（如测试账号、机器人账号等），然后再统计真实用户的数量。
from pyspark.sql.functions import countDistinct
# 模拟用户活跃数据
data = [
("user1", 10, False), # 正常用户
("user2", 1000, False), # 异常用户（行为次数过多）
("user3", 5, True), # 测试账号
("user4", 8, False), # 正常用户
("user5", 1500, False) # 异常用户
]
df = spark.createDataFrame(data, ["user_id", "action_count", "is_test"])
# 过滤异常值：行为次数在合理范围内且不是测试账号
filtered_df = df.filter(
(col("action_count") < 100) &
(col("is_test") == False)
)
# 统计唯一用户数
unique_users = filtered_df.agg(countDistinct("user_id").alias("unique_users"))
unique_users.show()
异常值过滤策略：
| 异常类型 | 过滤条件 | 说明 |
|---|---|---|
| 测试账号 | is_test == True |
标记为测试的账号 |
| 机器人行为 | action_count > 100 |
单日行为次数异常高 |
| 无效用户 | user_id is null |
用户ID为空 |
3. 场景三：清洗并合并多来源数据
当数据来自多个来源时，经常会出现同一实体在不同系统中有不同表示的情况。我们需要先标准化数据，再去重合并。
from pyspark.sql.functions import lower, trim, regexp_replace
# 模拟来自两个系统的用户数据
data_system1 = [
(" john.doe@example.com ", "John Doe", "New York"),
("alice.smith@example.com", "Alice Smith", "Boston"),
("bob.johnson@example.com", "Bob Johnson", "Chicago")
]
data_system2 = [
("JOHN.DOE@example.com", "John Doe", "NY"),
("alice.smith@example.com", "Alice Smith", "Boston, MA"),
("charlie.brown@example.com", "Charlie Brown", "Seattle")
]
df1 = spark.createDataFrame(data_system1, ["email", "name", "city"])
df2 = spark.createDataFrame(data_system2, ["email", "name", "city"])
# 数据标准化处理
def standardize(df):
return df.withColumn("email", lower(trim(regexp_replace(col("email"), "\\s+", "")))) \
.withColumn("name", trim(col("name"))) \
.withColumn("city", trim(col("city")))
df1_std = standardize(df1)
df2_std = standardize(df2)
# 合并并去重
combined_df = df1_std.union(df2_std).distinct()
combined_df.show()
数据标准化步骤：
- 去除前后空格（
trim） - 统一大小写（
lower） - 去除中间空格（
regexp_replace） - 合并后使用
distinct去重
4. 场景四：基于多条件的高级去重
在某些情况下，简单的 distinct 可能不够，我们需要基于多个条件的组合来判断记录是否重复。
from pyspark.sql.functions import concat_ws
# 模拟电商订单数据
data = [
(1, "user1", "2023-01-01", "productA", 2, 50.0),
(2, "user1", "2023-01-01", "productA", 2, 50.0), # 完全重复
(3, "user1", "2023-01-01", "productA", 1, 50.0), # 数量不同
(4, "user1", "2023-01-02", "productA", 2, 50.0), # 日期不同
(5, "user2", "2023-01-01", "productA", 2, 50.0) # 用户不同
]
df = spark.createDataFrame(data, ["order_id", "user_id", "date", "product", "quantity", "price"])
# 定义去重键：用户+日期+产品+价格
df_with_key = df.withColumn("dedup_key",
concat_ws("|", col("user_id"), col("date"), col("product"), col("price"))
)
# 使用窗口函数保留每组重复记录中的第一条
window = Window.partitionBy("dedup_key").orderBy("order_id")
result = df_with_key.withColumn("row_num", row_number().over(window)) \
.filter(col("row_num") == 1) \
.drop("row_num", "dedup_key")
result.show()
去重策略选择：
| 场景 | 去重键 | 说明 |
|---|---|---|
| 完全重复 | 所有字段 | 所有字段值都相同 |
| 业务重复 | 关键业务字段 | 如用户+产品+日期 |
| 部分重复 | 选择字段组合 | 如用户+产品+价格 |
5. 场景五：构建完整的数据清洗Pipeline
将上述技术组合起来，构建一个完整的数据清洗Pipeline，包含数据质量检查点。
from pyspark.sql.functions import when, count, lit
# 模拟原始数据
data = [
("user1", "2023-01-01", "productA", 2, 100.0),
("user1", "2023-01-01", "productA", 2, 100.0), # 完全重复
("user2", "2023-01-01", "productB", -1, 50.0), # 数量异常
(None, "2023-01-02", "productC", 1, 75.0), # 用户ID缺失
("user3", "2023-01-03", "productD", 3, None) # 价格缺失
]
df = spark.createDataFrame(data, ["user_id", "date", "product", "quantity", "price"])
# 数据质量检查
def data_quality_check(df):
# 计算各种数据质量问题计数
quality_metrics = df.agg(
count(when(col("user_id").isNull(), 1)).alias("null_user_ids"),
count(when(col("quantity") <= 0, 1)).alias("invalid_quantities"),
count(when(col("price").isNull(), 1)).alias("null_prices"),
count(lit(1)).alias("total_records")
)
return quality_metrics
# 打印数据质量问题报告
quality_report = data_quality_check(df)
quality_report.show()
# 数据清洗Pipeline
def clean_data(df):
# 1. 去除完全重复记录
df = df.distinct()
# 2. 过滤无效记录
df = df.filter(
col("user_id").isNotNull() &
(col("quantity") > 0) &
col("price").isNotNull()
)
# 3. 标准化日期格式
df = df.withColumn("date", to_date(col("date"), "yyyy-MM-dd"))
return df
# 执行清洗
cleaned_df = clean_data(df)
cleaned_df.show()
# 清洗后数据质量检查
clean_quality_report = data_quality_check(cleaned_df)
clean_quality_report.show()
数据清洗Pipeline步骤：
- 去重阶段 ：使用
distinct去除完全重复记录 - 过滤阶段 ：使用
filter去除无效记录 - 标准化阶段 ：统一日期、格式等
- 验证阶段 ：再次检查数据质量
最佳实践与性能优化
在实际应用中，数据清洗的性能至关重要。以下是几个优化建议：
-
分区策略 ：对于大数据集，先按关键字段分区可以提高去重效率
df.repartition("user_id", "date") -
缓存中间结果 ：对于需要多次使用的DataFrame，先缓存起来
df.cache() -
合理设置并行度 ：根据集群资源调整并行度
spark.conf.set("spark.sql.shuffle.partitions", "200") -
选择性去重 ：如果只需要对部分字段去重，使用
dropDuplicates而不是distinctdf.dropDuplicates(["user_id", "date"]) -
监控数据质量 ：建立数据质量监控指标，定期检查
| 指标 | 计算公式 | 阈值 |
|---|---|---|
| 重复率 | 重复记录数/总记录数 | <1% |
| 缺失率 | 缺失字段数/总字段数 | <5% |
| 异常值率 | 异常记录数/总记录数 | <2% |
数据清洗是大数据分析中不可或缺的一环。通过合理组合 filter 和 distinct 等基础算子，我们可以解决大多数数据质量问题。在实际项目中，建议根据具体业务需求调整清洗策略，并建立完善的数据质量监控体系。
更多推荐
8
0- 0
已为社区贡献44条内容
所有评论(0)