PySpark大表空值、空白值及可疑值统计性能优化问询

核心思路 #

将所有列的统计逻辑合并为一组表达式，通过单次全表扫描完成所有计算，彻底避免多次触发Job重复加载数据的问题。利用Spark的agg方法批量传入所有统计表达式，让Spark引擎自动优化执行计划，最大化复用数据扫描结果。

需统计的核心项 #

针对每列统一计算以下指标：

• 空值（null）数量
• 非空值数量
• 空白值（空字符串、仅含空格的字符串）数量
• 唯一值数量
• 指定可疑值（如none、null、nan等）的出现次数

示例代码 #

1. 定义列统计表达式生成函数 #

from pyspark.sql import functions as F

def get_col_stats_exprs(col_name, suspicious_vals=["none", "null", "nan"]):
    # 空值统计
    null_cnt = F.count(F.when(F.col(col_name).isNull(), 1)).alias(f"{col_name}_null_cnt")
    # 非空值统计
    non_null_cnt = F.count(col_name).alias(f"{col_name}_non_null_cnt")
    # 空白值统计（兼容非字符串列，将非字符串类型转为字符串后判断）
    blank_cnt = F.count(
        F.when(
            (F.col(col_name).cast("string").trim() == "") | (F.col(col_name).isNull()),
            1
        )
    ).alias(f"{col_name}_blank_cnt")
    # 唯一值数量
    distinct_cnt = F.countDistinct(col_name).alias(f"{col_name}_distinct_cnt")
    # 可疑值统计（区分字符串/数值类型，避免类型转换错误）
    suspicious_exprs = []
    for val in suspicious_vals:
        if isinstance(val, str):
            match = F.upper(F.col(col_name).cast("string")) == F.upper(F.lit(val))
        else:
            match = F.col(col_name) == F.lit(val)
        suspicious_exprs.append(
            F.count(F.when(match, 1)).alias(f"{col_name}_suspicious_{val}_cnt")
        )
    # 合并所有表达式
    return [null_cnt, non_null_cnt, blank_cnt, distinct_cnt] + suspicious_exprs

2. 批量生成表达式并执行统计 #

# 假设df为目标DataFrame
all_cols = df.columns
# 生成所有列的统计表达式
total_stats_exprs = []
for col in all_cols:
    total_stats_exprs.extend(get_col_stats_exprs(col, suspicious_vals=["none", "null", "nan"]))
# 单次执行所有统计逻辑
stats_result = df.agg(*total_stats_exprs).collect()[0]
# 转换为字典，方便后续查询
stats_dict = stats_result.asDict()

3. 结果查询示例 #

# 查看某一列的完整统计结果
target_col = "order_id"
print(f"列 {target_col} 统计结果：")
print(f"空值数量: {stats_dict[f'{target_col}_null_cnt']}")
print(f"非空值数量: {stats_dict[f'{target_col}_non_null_cnt']}")
print(f"空白值数量: {stats_dict[f'{target_col}_blank_cnt']}")
print(f"唯一值数量: {stats_dict[f'{target_col}_distinct_cnt']}")
print(f"可疑值nan出现次数: {stats_dict[f'{target_col}_suspicious_nan_cnt']}")

性能优化补充点 #

• 减少Shuffle开销：如果对唯一值精度要求不高，可用approx_count_distinct替代countDistinct，大幅降低Shuffle数据量：

  distinct_cnt = F.approx_count_distinct(col_name, rsd=0.05).alias(f"{col_name}_distinct_cnt")
  

  其中rsd为相对标准偏差，值越小精度越高，默认0.05。
• 类型适配优化：针对数值型列跳过字符串转换逻辑，可单独编写分支进一步减少计算开销。
• 分区调整：若原表分区不合理，可先通过repartition调整分区数，让Spark并行度匹配集群资源。

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