PySpark DataFrame多列空值与重复值统计优化方案咨询

原始DataFrame #

data = [
    ("James","CA",None), (None,"AC",None),
    ("Ram","AC",200.0), ("Ram",None,None)
]
df = spark.createDataFrame(data,["name","state","number"])
df.show()

输出结果：

+-----+-----+------+
| name|state|number|
+-----+-----+------+
|James|   CA|  null|
| null|   AC|  null|
|  Ram|   AC| 200.0|
|  Ram| null|  null|
+-----+-----+------+

需求说明 #

• 统计每列的null数量，并将所有列的空值总数存入变量
• 统计每列的重复值数量，并将所有列的重复值总数存入变量

现有实现 #

空值统计 #

from pyspark.sql.functions import col,isnan, when, count
df_null = df.select([count(when(col(c).isNull(), c)).alias(c) for c in df.columns])
df_null.show()

输出结果：

+-----+-----+------+
| name|state|number|
+-----+-----+------+
|    1|    1|     3|
+-----+-----+------+

手动提取并求和：

n_1 = df_null.collect()[0][0]
s_1 = df_null.collect()[0][1]   
nu_1 = df_null.collect()[0][2]

null_count = n_1 + s_1 + nu_1

重复值统计 #

list_1 = ['name']
df_1 = df.groupby(list_1).count().where('count > 1')
# 输出：
# +-----+-----+
# | name|count|
# +-----+-----+
# |  Ram|    2|
# +-----+-----+

list_2 = ['state']
df_2 = df.groupby(list_2).count().where('count > 1')
# 输出：
# +------+-----+
# | state|count|
# +------+-----+
# |    AC|    2|
# +------+-----+

list_df1 = df_1.collect()[0][1]
list_df2 = df_2.collect()[0][1]

dup_count = list_df1 + list_df2

优化实现 #

空值统计优化 #

利用Row.asDict()将统计结果转为字典，直接求和所有值，无需硬编码列索引，扩展性更强：

from pyspark.sql.functions import col, when, count

# 统计每列空值数
df_null = df.select([count(when(col(c).isNull(), c)).alias(c) for c in df.columns])
# 转为字典后求和
null_stats = df_null.collect()[0].asDict()
total_null = sum(null_stats.values())

# 打印每列空值数和总数
print("每列空值数:", null_stats)
print("总空值数:", total_null)

重复值统计优化 #

通过循环遍历所有列，统一处理分组统计逻辑，避免重复代码，支持任意列数：

total_dup = 0
column_dup_stats = {}

for col_name in df.columns:
    # 分组统计当前列的出现次数，筛选出现次数>1的分组
    dup_group = df.groupBy(col_name).count().filter("count > 1")
    # 累加当前列的重复count总和（若无重复则加0）
    col_dup_sum = dup_group.agg({"count": "sum"}).collect()[0][0] or 0
    column_dup_stats[col_name] = col_dup_sum
    total_dup += col_dup_sum

# 打印每列重复值统计和总数
print("每列重复值count总和:", column_dup_stats)
print("总重复值count总和:", total_dup)

注：如果需求是统计重复的条目数（即每个重复组中除首次出现外的数量，例如Ram出现2次则计1次重复），只需将col_dup_sum的计算改为：

col_dup_sum = dup_group.agg({"count": lambda x: sum(i-1 for i in x)}).collect()[0][0] or 0

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