PySpark 计算每列“缺失”的百分比

Answer 1

假设您不考虑缺失值计数的几列（这里我假设您的列id 不应包含缺失值），您可以使用以下代码

import pyspark.sql.functions as F

# select columns in which you want to check for missing values
relevant_columns = [c for c in df.columns if c != 'id']

# number of total records
n_records = df.count()

# percentage of rows with all missings in relevant_columns
my_perc = df \
  .select((F.lit(len(relevant_columns)) - (sum(df[c].isNull().cast('int') for c in relevant_columns))).alias('n')) \
  .filter(F.col('n') == 0) \
  .count() / n_records * 100

print(my_perc)
# 25.0

Answer 2

另一种方法是创建一个自定义函数 - calc_null_percent 利用来自 Spark 和 Pandas 的两全其美

自定义 func 将包含 total_count 和 null_count 分别对应于每一列

数据准备

from pyspark import SparkContext
from pyspark.sql import SQLContext
from functools import reduce
import pyspark.sql.functions as F
import pandas as pd
import numpy as np

from io import StringIO

sc = SparkContext.getOrCreate()
sql = SQLContext(sc)

input_str = """
1,0,null,
1,null,0,
null,1,0,
1,0,0,
1,0,0,
null,0,1,
1,1,0,
1,1,null,
null,1,0
""".split(',')

input_values = list(map(lambda x: x.strip() if x.strip() != 'null' else None, input_str))

cols = list(map(lambda x: x.strip() if x.strip() != 'null' else None, "col1,col2,col3".split(',')))
            
n = len(input_values)

n_col = 3

input_list = [tuple(input_values[i:i+n_col]) for i in range(0,n,n_col)]

sparkDF = sql.createDataFrame(input_list, cols)

sparkDF.show()

+----+----+----+
|col1|col2|col3|
+----+----+----+
|   1|   0|null|
|   1|null|   0|
|null|   1|   0|
|   1|   0|   0|
|   1|   0|   0|
|null|   0|   1|
|   1|   1|   0|
|   1|   1|null|
|null|   1|   0|
+----+----+----+

自定义函数

def calc_null_percent(spark_df,sort=True):
    
    pd_col_count = spark_df.select([F.count(F.col(c)).alias(c) 
                                        for (c,c_type) in spark_df.dtypes]
                                  ).toPandas().T.reset_index().rename(columns={0: 'total_count'
                                                                               ,'index':'column'})
    
    pd_col_null_count = spark_df.select([F.sum(F.when(F.isnan(c) | F.isnull(c),1).otherwise(0)).alias(c)
                                         for (c,c_type) in spark_df.dtypes]
                                       ).toPandas().T.reset_index().rename(columns={0: 'null_count'
                                                                                    ,'index':'column'})
    
    final_df = pd.merge(pd_col_count,pd_col_null_count,on=['column'])
    
    final_df['null_percentage'] = final_df['null_count'] * 100 / final_df['total_count']

    if len(final_df) == 0:
        print("There are no any missing values!")
        return None

    return final_df

nullStatsDF = sql.createDataFrame(calc_null_percent(sparkDF))

nullStatsDF.show()

+------+-----------+----------+------------------+
|column|total_count|null_count|   null_percentage|
+------+-----------+----------+------------------+
|  col1|          6|         3|              50.0|
|  col2|          8|         1|              12.5|
|  col3|          7|         2|28.571428571428573|
+------+-----------+----------+------------------+