在PySpark数据框中删除包含特定值的行

Question

我有一个类似于以下的pyspark数据框：

A    B      C 1    NA     9 4    2       5 6    4       2 5    1    NA

我想要删除包含值 "NA" 的行。在这种情况下是第一行和最后一行。如何使用Python和Spark实现这一操作？

根据评论更新：

寻找一个解决方案，可以删除任何一个列中具有字符串 "NA" 的行。

- jason_1093

---

NA是什么意思？它是否表示缺失值？或者在您的DataFrame中是这样的吗？如果是这样，那么您的B列将是字符串！请确认。 - cph_sto

NA不是缺失值，它是一个字符串关键字。我想删除所有包含字符串“NA”的行。 - jason_1093

此外，NA也可能出现在另一列中，不一定是B列，因此该行也应被删除。 - jason_1093

是的，因为存在“NA”，所以Spark将它们标记为字符串。我想要删除“NA”，以便可以将列标记为整数。一种方法是在所有地方用0替换“NA”。我无法使用正确的语法实现它。 - jason_1093

你应该添加一个最小的示例以及你已经尝试过的内容。一个简单的过滤器就可以完成这项工作。 - Yeikel

Answer 1

只需使用数据框 筛选 表达式：

l = [('1','NA','9')
    ,('4','2', '5')
    ,('6','4','2')
    ,('5','NA','1')]
df = spark.createDataFrame(l,['A','B','C'])
#The following command requires that the checked columns are strings!
df = df.filter((df.A != 'NA') & (df.B != 'NA') & (df.C != 'NA'))
df.show()

+---+---+---+ 
|  A|  B|  C| 
+---+---+---+ 
|  4|  2|  5| 
|  6|  4|  2| 
+---+---+---+

@bluephantom：如果你有数百列的情况，只需通过列表推导式生成一个字符串表达式：

#In my example are columns need to be checked
listOfRelevantStringColumns = df.columns
expr = ' and '.join('(%s != "NA")' % col_name for col_name in listOfRelevantStringColumns)
df.filter(expr).show()

- cronoik

---

如果有数百根柱子，该怎么办？ - thebluephantom

我在思考的是如何让它工作并在找到第一个匹配项后停止？我正在从Scala方面研究它。 - thebluephantom

你想在第一行匹配或第一列匹配后停止吗？也许开一个新问题来讨论这个问题会更有意义。 - cronoik

我正在尝试使用if语句，并将其作为Scala答案呈现。无法找到类似于搜索所有列的数组搜索功能。 - thebluephantom

我发布了一个中间解决方案作为替代。 - thebluephantom

Answer 2

如果你想要删除这一行

df = df.filter((df.A != 'NA') | (df.B != 'NA'))

有时候我们需要用平均值（在数值列的情况下）或者众数（在分类列的情况下）来替换。为此，您需要添加一个与原始列名相同的列，即“A”。

from pyspark.sql.functions import mean,col,when,count
df=df.withColumn("A",when(df.A=="NA",mean(df.A)).otherwise(df.A))

- Ghias Ali

---

应该使用 & 而不是 |。 - Yash Khasgiwala

如果行中两列都有“NA”值，则使用“&”连接符，如果其中任意一列有“NA”值，则使用“|”过滤符。 - Ghias Ali

Answer 3

在Scala中，我做了不同的事情，但是使用pyspark得到了这个结果。虽然不是我最喜欢的答案，但这是因为我对pyspark的知识较少。在Scala中，事情似乎更容易。与数组不同，没有全局匹配所有列的匹配可以在找到一个后停止。根据列数动态变化。

假设数据中没有~~作为数据的一部分，可以将其拆分为数组，但决定在此处不这样做。使用None而不是NA。

from pyspark.sql import functions as f

data = [(1,    None,    4,    None),
        (2,    'c',     3,    'd'),
        (None, None,    None, None),
        (3,    None,    None, 'z')]
df = spark.createDataFrame(data, ['k', 'v1', 'v2', 'v3'])

columns = df.columns
columns_Count = len(df.columns)

# colCompare is String
df2 = df.select(df['*'], f.concat_ws('~~', *columns).alias('colCompare') )
df3 = df2.filter(f.size(f.split(f.col("colCompare"), r"~~"))  == columns_Count).drop("colCompare")
df3.show()

返回：

+---+---+---+---+
|  k| v1| v2| v3|
+---+---+---+---+
|  2|  c|  3|  d|
+---+---+---+---+

- thebluephantom