Pyspark - 计算每个数据框列中的空值数量

Posted 2023-04-15

【中文标题】Pyspark - 计算每个数据框列中的空值数量

【英文标题】：Pyspark - Calculate number of null values in each dataframe column

【发布时间】：2018-09-12 16:08:55

【问题描述】：

我有一个包含许多列的数据框。我的目标是生成一个数据框，其中列出了每个列名以及该列中空值的数量。

例子：

+-------------+-------------+
| Column_Name | NULL_Values |
+-------------+-------------+
|  Column_1   |      15     |
|  Column_2   |      56     |
|  Column_3   |      18     |
|     ...     |     ...     |
+-------------+-------------+

我已经设法获得了 ONE 列的空值数量，如下所示：

df.agg(F.count(F.when(F.isnull(c), c)).alias('NULL_Count'))

其中c 是数据框中的一列。但是，它不显示列的名称。输出是：

+------------+
| NULL_Count |
+------------+
|     15     |
+------------+

有什么想法吗？

【参考方案1】：

您可以使用列表推导循环遍历 agg 中的所有列，并使用 alias 重命名输出列：

import pyspark.sql.functions as F

df_agg = df.agg(*[F.count(F.when(F.isnull(c), c)).alias(c) for c in df.columns])

但是，这将在一行中返回结果，如下所示：

df_agg.show()
#+--------+--------+--------+
#|Column_1|Column_2|Column_3|
#+--------+--------+--------+
#|      15|      56|      18|
#+--------+--------+--------+

如果您希望将结果放在一列中，您可以使用 functools.reduce 合并来自 df_agg 的每一列，如下所示：

from functools import reduce
df_agg_col = reduce(
    lambda a, b: a.union(b),
    (
        df_agg.select(F.lit(c).alias("Column_Name"), F.col(c).alias("NULL_Count")) 
        for c in df_agg.columns
    )
)
df_agg_col.show()
#+-----------+----------+
#|Column_Name|NULL_Count|
#+-----------+----------+
#|   Column_1|        15|
#|   Column_2|        56|
#|   Column_3|        18|
#+-----------+----------+

或者您可以跳过创建df_agg 的中间步骤并执行以下操作：

df_agg_col = reduce(
    lambda a, b: a.union(b),
    (
        df.agg(
            F.count(F.when(F.isnull(c), c)).alias('NULL_Count')
        ).select(F.lit(c).alias("Column_Name"), "NULL_Count")
        for c in df.columns
    )
)

【讨论】：

完美运行，谢谢！您能解释一下星号在您的第一个聚合函数中的作用吗？

@LEJ * 代表argument unpacking。此语法“解包”列表的内容，以便它们可以作为参数传递给函数。

【参考方案2】：

Scala 替代品可能是

case class Test(id:Int, weight:Option[Int], age:Int, gender: Option[String])

val df1 = Seq(Test(1, Some(100), 23, Some("Male")), Test(2, None, 25, None), Test(3, None, 33, Some("Female"))).toDF()

df1.show()

+---+------+---+------+
| id|weight|age|gender|
+---+------+---+------+
|  1|   100| 23|  Male|
|  2|  null| 25|  null|
|  3|  null| 33|Female|
+---+------+---+------+

val s = df1.columns.map(c => sum(col(c).isNull.cast("integer")).alias(c))

val df2 = df1.agg(s.head, s.tail:_*)

val t = df2.columns.map(c => df2.select(lit(c).alias("col_name"), col(c).alias("null_count")))

val df_agg_col = t.reduce((df1, df2) => df1.union(df2))

df_agg_col.show()