PySpark删除常数列时如何保留含空值与单一非空值的列

现有PySpark常数列识别方案的核心偏差来自countDistinct()的原生逻辑：该函数统计时完全忽略null值，导致列的实际取值多样性被低估：

• 当列仅包含null和1种非空值时，countDistinct()返回值为1，会被传统“去重值数<2就删除”的逻辑误判为常数列
• 传统替换null为占位值的方案需要额外选定不存在的占位值，存在占位值和真实业务值冲突的风险，且需要额外的写操作遍历全量数据，实现冗余。

不需要修改原数据，只需要将「列中是否存在null」作为独立的统计维度，计算列的等效去重值总数，以此判定真常数列：

• 真常数列的判定标准：等效去重值总数 ≤1
• 等效去重值总数的计算规则：基础值取列的非空值去重数量（直接用原生countDistinct()即可）；如果该列存在null值，则加1（将null视为一个独立的取值类别）

按照这个规则的判定结果完全匹配需求：

• 全列均为同一非空值、无null：等效去重数=1 → 判定为常数列
• 全列均为null：等效去重数=1（非空去重数=0 + null维度补1） → 判定为常数列
• 列仅含1种非空值+若干null：等效去重数=2 → 不判定为常数列，保留

首先导入依赖：

from pyspark.sql import functions as F

封装常数列识别函数：

def detect_constant_columns(input_df, max_effective_distinct=1):
    # 一次聚合拿到所有列的统计指标
    agg_expressions = []
    # 非空值去重数统计
    for col_name in input_df.columns:
        agg_expressions.append(
            F.countDistinct(F.col(col_name)).alias(f"nonnull_distinct_{col_name}")
        )
    # null值存在性统计
    for col_name in input_df.columns:
        agg_expressions.append(
            F.sum(F.when(F.col(col_name).isNull(), 1).otherwise(0)).alias(f"null_cnt_{col_name}")
        )
    # 拉取单行聚合结果（聚合结果仅1行，无大内存风险）
    agg_result = input_df.agg(*agg_expressions).collect()[0].asDict()

    constant_cols = []
    for col_name in input_df.columns:
        nonnull_distinct = agg_result[f"nonnull_distinct_{col_name}"]
        has_null = agg_result[f"null_cnt_{col_name}"] > 0
        # 计算等效去重总数
        effective_distinct = nonnull_distinct + (1 if has_null else 0)
        if effective_distinct <= max_effective_distinct:
            constant_cols.append(col_name)
    return constant_cols

使用方式：

# 识别所有真常数列
constant_cols = detect_constant_columns(your_dataframe)
# 从原DataFrame中删除识别出的常数列
cleaned_df = your_dataframe.drop(*constant_cols)

提示：如果业务场景中需要保留全空列、仅删除全为同一非空值的常数列，只需要在判定时增加过滤条件，跳过nonnull_distinct=0的列即可，逻辑调整灵活。

• 完全无侵入：不需要对原数据做null替换、值改写等操作，没有占位值与业务值冲突的隐患
• 性能更优：所有统计通过一次全表聚合完成，没有冗余的全量扫描开销
• 判定逻辑精准：既可以识别删除无业务价值的纯常数列、全空列，也不会误删存在少量null+单一非空值的有效列

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