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.doeexample.com , John Doe, New York), (alice.smithexample.com, Alice Smith, Boston), (bob.johnsonexample.com, Bob Johnson, Chicago) ] data_system2 [ (JOHN.DOEexample.com, John Doe, NY), (alice.smithexample.com, Alice Smith, Boston, MA), (charlie.brownexample.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等基础算子我们可以解决大多数数据质量问题。在实际项目中建议根据具体业务需求调整清洗策略并建立完善的数据质量监控体系。