是否可以通过 Spark 中的值的总和来过滤列？

Posted 2023-04-15

【中文标题】是否可以通过 Spark 中的值的总和来过滤列？

【英文标题】：Is it possible to filter columns by the sum of their values in Spark?

【发布时间】：2020-09-09 14:27:36

【问题描述】：

我正在使用 PySpark 加载一个稀疏表，我想删除列中所有值的总和高于阈值的所有列。

例如下表列值的总和：

+---+---+---+---+---+---+
|  a|  b|  c|  d|  e|  f|
+---+---+---+---+---+---+
|  1|  0|  1|  1|  0|  0|
|  1|  1|  0|  0|  0|  0|
|  1|  0|  0|  1|  1|  1|
|  1|  0|  0|  1|  1|  1|
|  1|  1|  0|  0|  1|  0|
|  0|  0|  1|  0|  1|  0|
+---+---+---+---+---+---+

是 5、2、2、3、4 和 2。过滤总和 >= 3 的所有列应输出此表：

+---+---+---+
|  a|  d|  e|
+---+---+---+
|  1|  1|  0|
|  1|  0|  0|
|  1|  1|  1|
|  1|  1|  1|
|  1|  0|  1|
|  0|  0|  1|
+---+---+---+

我尝试了许多不同的解决方案，但均未成功。 df.groupBy().sum() 给了我列值的总和，所以我正在搜索如何过滤具有阈值的值并仅从原始数据框中获取剩余的列。

由于不仅有 6 列而且有数千列，因此我正在寻找一种可扩展的解决方案，我不必输入每个列的名称。感谢您的帮助！

【参考方案1】：

您可以通过collect（或first）步骤来做到这一点。

from pyspark.sql import functions as F

sum_result = df.groupBy().agg(*(F.sum(col).alias(col) for col in df.columns)).first()

filtered_df = df.select(
    *(col for col, value in sum_result.asDict().items() if value >= 3)
)

filtered_df.show()
+---+---+---+
|  a|  d|  e|
+---+---+---+
|  1|  1|  0|
|  1|  0|  0|
|  1|  1|  1|
|  1|  1|  1|
|  1|  0|  1|
|  0|  0|  1|
+---+---+---+

【讨论】：

非常有用，它真的拯救了我的一天！此外，我在F.sum 中添加了一个额外的F.when(col(c).isNotNull(),1))，因为值> 1，我只想计算！= 0 的出现次数。总体而言，这似乎是一个优雅的解决方案。

@ArnoXf 我不确定是否了解when 部分的需要。最后，您只想保留至少一个非空值的列，不是吗？

不完全是。真实数据集有时包括 2s 或 3s 而不仅仅是 0s（真实数据集中的null）和 1s。我想计算 one or more 的出现次数，这就是为什么我要告诉 sum 在它不为空时将所有内容视为 1。