计算 PySpark 中列中的非空值

Question

我有一个包含空值的数据帧：

from pyspark.sql import functions as F
df = spark.createDataFrame(
[(125, '2012-10-10', 'tv'),
(20, '2012-10-10', 'phone'),
(40, '2012-10-10', 'tv'),
(None, '2012-10-10', 'tv')],
["Sales", "date", "product"]
)

我需要计算"销售额"列中的非空值。

我尝试了 3 种方法。

第一个我做对了：

df.where(F.col("sales").isNotNull()).groupBy('product')
.agg((F.count(F.col("Sales")).alias("sales_count"))).show()
# product   | sales_count
# phone     |  1
# tv        |  2

第二个，这是不正确的：

df.groupBy('product')
.agg((F.count(F.col("Sales").isNotNull()).alias("sales_count"))).show()
# product   | sales_count
# phone     |  1
# tv        |  3

第三个，我得到了错误：

df.groupBy('product')
.agg((F.col("Sales").isNotNull().count()).alias("sales_count")).show()

类型错误："列"对象不可调用

什么可能导致第二种和第三种方法出错？

Answer 1

您的第一次尝试是在执行聚合之前过滤掉Sales列中带有null的行。因此，它为您提供了正确的结果。

但是使用第二个代码

df.groupBy('product') 
.agg((F.count(F.col("Sales").isNotNull()).alias("sales_count"))).show()

您还没有filtered出来，而是对整个数据集进行了aggregation。如果你仔细分析F.col("Sales").isNotNull()会给你布尔列，即true和false.因此，F.count(F.col("Sales").isNotNull())只是计算分组数据集中的布尔值，如果您创建一个新列，如下所示，这一点很明显。

df.withColumn("isNotNull", F.col("Sales").isNotNull()).show()

这会给你

+-----+----------+-------+---------+
|Sales|      date|product|isNotNull|
+-----+----------+-------+---------+
|  125|2012-10-10|     tv|     true|
|   20|2012-10-10|  phone|     true|
|   40|2012-10-10|     tv|     true|
| null|2012-10-10|     tv|    false|
+-----+----------+-------+---------+

因此，第二次尝试的计数是正确的。

对于第三次尝试，.count()是不能在聚合转换中使用的操作。只有返回 dataTypeColumn函数才能在.agg()中使用，它们可以是内置函数、UDF 或您自己的函数。

Answer 2

有一个更简单的方法：

>>> df.groupBy("product").agg({"Sales":"count"}).show()
+-------+------------+
|product|count(Sales)|
+-------+------------+
|  phone|           1|
|     tv|           2|
+-------+------------+

Answer 3

计算非空值

• 仅适用于每个字符串和数字列：

  df.summary("count").show()
  # +-------+-----+----+-------+
  # |summary|Sales|date|product|
  # +-------+-----+----+-------+
  # |  count|    3|   4|      4|
  # +-------+-----+----+-------+
  

• 对于任何类型的每一列：

  df.agg(*[F.count(c).alias(c) for c in df.columns]).show()
  # +-----+----+-------+
  # |Sales|date|product|
  # +-----+----+-------+
  # |    3|   4|      4|
  # +-----+----+-------+