如何使用 PySpark 在数据框中删除基于多个过滤器的列？

Question

我有一个单元格可以具有的有效值列表。如果一列中的一个单元格无效，我需要删除整列。我知道有在特定列中删除行的答案，但在这里我将删除整个列，即使其中的一个单元格无效。有效/无效的条件是一个单元格只能有三个值：['Messi', 'Ronaldo', 'Virgil']

我尝试阅读有关过滤的内容，但我看到的只是过滤列并删除行。例如在this 问题中。我还读到应该避免在 Spark 中进行过多的扫描和洗牌，我同意这一点。

我不仅关注代码解决方案，还关注 PySpark 提供的现成代码。我希望它不会超出 SO 答案的范围。

对于以下输入数据框：

| Column 1      | Column 2      | Column 3      | Column 4      | Column 5      |
| --------------| --------------| --------------| --------------| --------------|
|  Ronaldo      | Salah         |  Messi        |               |Salah          |
|  Ronaldo      | Messi         |  Virgil       |  Messi        | null          |
|  Ronaldo      | Ronaldo       |  Messi        |  Ronaldo      | null          |

我希望得到以下输出：

| Column 1      | Column 2      |
| --------------| --------------| 
|  Ronaldo      | Messi         |
|  Ronaldo      | Virgil        |
|  Ronaldo      | Messi         |

Answer 1

我不仅关注代码解决方案，还关注 PySpark 提供的现成代码。

不幸的是，Spark 旨在逐行并行运行。过滤掉列不是“现成代码”解决方案。

不过，您可以采取以下一种方法：

首先收集每一列中无效元素的计数。

from pyspark.sql.functions import col, lit, sum as _sum, when

valid = ['Messi', 'Ronaldo', 'Virgil']
invalid_counts = df.select(
    *[_sum(when(col(c).isin(valid), lit(0)).otherwise(lit(1))).alias(c) for c in df.columns]
).collect()
print(invalid_counts)
#[Row(Column 1=0, Column 2=1, Column 3=0, Column 4=1, Column 5=3)]

此输出将是一个只有一个元素的列表。您可以遍历此元素中的项目以查找要保留的列。

valid_columns = [k for k,v in invalid_counts[0].asDict().items() if v == 0]
print(valid_columns)
#['Column 3', 'Column 1']

现在只需从原始 DataFrame 中选择这些列。如果要保持原始列顺序，可以先使用list.index 对valid_columns 进行排序。

valid_columns = sorted(valid_columns, key=df.columns.index)
df.select(valid_columns).show()
#+--------+--------+
#|Column 1|Column 3|
#+--------+--------+
#| Ronaldo|   Messi|
#| Ronaldo|  Virgil|
#| Ronaldo|   Messi|
#+--------+--------+