PySpark分组聚合与去重后count结果不一致问题求助

核心原因 #

这种count结果不稳定的情况主要由分布式shuffle过程中的数据一致性问题导致，小数据集无异常是因为未触发大规模跨节点数据传输，分布式环境下的隐性问题未暴露。集群配置不合理可能加剧该问题，但并非直接诱因。

排查与修复步骤 #

1. 校验分组/去重键的有效性 #

• 数据类型问题：若col1/col2或去重键为浮点型（float/double），精度丢失会导致逻辑相同的值被判定为不同分组。转换为高精度Decimal类型：

  from pyspark.sql.types import DecimalType
  df = df.withColumn("col1", df["col1"].cast(DecimalType(18, 6)))
  df = df.withColumn("col2", df["col2"].cast(DecimalType(18, 6)))
  

• 空值/脏数据问题：隐性空值（如NaN、空字符串）会破坏分组/去重逻辑，提前过滤清理：

  # 分组前清理
  df_clean = df.filter(df.col1.isNotNull() & df.col2.isNotNull())
  # 去重前清理
  df_dedup_clean = df_clean.filter(df_clean.col3.isNotNull() & df_clean.col4.isNotNull() & df_clean.col5.isNotNull() & df_clean.col6.isNotNull() & df_clean.col7.isNotNull())
  

2. 优化Shuffle相关配置 #

修改Spark参数，强制shuffle过程的一致性：

• 调整spark.sql.shuffle.partitions至总核心数的2-3倍（集群总核心114*5=570，建议设为1140），避免分区碎片化：

  spark.conf.set("spark.sql.shuffle.partitions", "1140")
  

• 启用shuffle一致性分区：

  spark.conf.set("spark.sql.shuffle.consistentPartitioning.enabled", "true")
  

• 关闭哈希分区随机性（Spark 3.x默认开启，强制统一哈希逻辑）：

  spark.conf.set("spark.sql.execution.arrow.fallback.enabled", "false")
  

3. 替换dropDuplicates为显式分组去重 #

dropDuplicates的底层实现依赖shuffle，显式使用groupBy+agg可更稳定地控制去重逻辑：

from pyspark.sql import functions as F
df_dedup = df_clean.groupBy("col3","col4","col5","col6","col7").agg(F.first("*").alias("row_data")).select("row_data.*")

4. 调整集群并行度配置 #

当前spark.default.parallelism=1700远大于总核心数570，任务碎片化会增加shuffle异常概率，调整为总核心数的2倍：

spark.conf.set("spark.default.parallelism", "1140")

5. 验证数据完整性 #

• 使用exceptAll对比两次运行的结果差异，定位异常行：

  df_run1 = ... # 第一次运行结果
  df_run2 = ... # 第二次运行结果
  diff_rows = df_run1.exceptAll(df_run2).union(df_run2.exceptAll(df_run1))
  diff_rows.show(100, truncate=False)
  

• 检查Parquet文件schema一致性，避免损坏文件导致的读取异常：

  spark.read.parquet("your-path").printSchema()
  

总结 #

通过规范键的数据类型、优化shuffle配置、调整并行度，可解决分布式环境下的count结果不稳定问题。核心是消除shuffle过程中哈希计算、数据分区的不确定性。

内容的提问来源于stack exchange，提问作者Atul G