PySpark空值处理函数fillnulls性能优化及最佳实践咨询

• 合并统计操作，减少全表扫描：避免对每个列单独计算众数/均值导致的多次数据集扫描。通过一次agg操作批量统计所有需要的指标：数值列的均值、非空值计数；字符串列的众数、总记录数，减少重复shuffle和IO开销。
  示例代码：

  from pyspark.sql import functions as F
  
  # 批量统计数值列的均值与非空计数
  num_stats = df.select([
      F.mean(col).alias(f"{col}_mean"),
      F.count(col).alias(f"{col}_count")
      for col in numeric_cols
  ]).collect()[0]
  
  # 批量统计字符串列的众数与总记录数
  str_stats = df.select([
      F.mode(col).alias(f"{col}_mode"),
      F.count(col).alias(f"{col}_total")
      for col in string_cols
  ]).collect()[0]
  
  # 预计算字符串列各值的出现次数，转为广播变量
  str_value_counts = {col: df.groupBy(col).count().collectAsMap() for col in string_cols}
  str_counts_broadcast = spark.sparkContext.broadcast(str_value_counts)
  

• 用广播变量传递低频值映射：将字符串列的低频值列表封装为广播变量，避免每个Task重复拉取数据，降低网络传输成本。
• 替换自定义UDF为内置函数组合：Python UDF性能远低于Spark内置JVM实现函数，改用when、coalesce等内置函数完成空值填充与低频值替换。
  示例代码：

  for col in string_cols:
      mode_val = str_stats[f"{col}_mode"]
      value_counts = str_counts_broadcast.value[col]
      low_freq_vals = [k for k, v in value_counts.items() if v < 100]
      
      df = df.withColumn(col,
          F.when(F.col(col).isNull(), mode_val)
           .when(F.col(col).isin(low_freq_vals), "other")
           .otherwise(F.col(col))
      )
  

• 调整Spark并行度：根据数据集大小修改spark.sql.shuffle.partitions（默认200），建议按每1GB数据对应1-2个分区的比例设置，避免分区过多/过少导致的资源浪费。
• 提前拦截无效列：处理前先检查数值列的非空值计数，若为0直接抛出异常，跳过后续无效计算。

• 优先使用内置统计函数：mean、mode、count等内置函数经过分布式优化，比手动实现的统计逻辑效率高数倍。
• 合理使用缓存：对需要重复访问的中间结果调用cache()或persist()，但避免缓存超大数据集引发内存溢出。
• 用近似统计替代精确统计：对精度要求不高的场景，用approx_count_distinct代替countDistinct、approxQuantile代替精确分位数计算，大幅降低计算资源消耗。
• 分区优化：按数据访问模式（如日期、核心类别列）分区，后续处理时仅扫描目标分区，减少数据处理量。
• 避免无意义shuffle：用broadcast join替代普通join，统计时优先在分区内做局部聚合再合并结果，减少跨节点数据传输。

• Spark ML Imputer组件：针对数值列，Imputer类支持批量填充均值/中位数，是分布式优化实现，性能优于手动计算填充。
  示例代码：

  from pyspark.ml.feature import Imputer
  
  imputer = Imputer(
      inputCols=numeric_cols,
      outputCols=[f"{col}_imputed" for col in numeric_cols],
      strategy="mean"
  )
  df_imputed = imputer.fit(df).transform(df)
  

• Pipeline整合全流程：将字符串列的众数填充、低频值替换与数值列的均值填充整合为ML Pipeline，利用Spark分布式处理能力，代码更简洁且性能更稳定。
• 批量fillna操作：提前统计好所有列的填充值后，用df.fillna()批量填充空值，比逐列处理效率更高：

  fill_map = {col: num_stats[f"{col}_mean"] for col in numeric_cols}
  fill_map.update({col: str_stats[f"{col}_mode"] for col in string_cols})
  df_filled = df.fillna(fill_map)
  # 后续单独处理字符串列低频值替换
  

内容的提问来源于stack exchange，提问作者Shaked Nave