我有一个单元格可以具有的有效值列表。如果一列中的一个单元格无效，我需要删除整列。我知道在特定列中有删除行的答案，但在这里我要删除整列，即使其中的一个单元格无效也是如此。有效/无效的条件是一个单元格只能有三个值:['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 |
请您参考如下方法:
I am not only looking at the code solution but more on the off-the-shelf code provided from 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|
#+--------+--------+
