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

【问题标题】：Drop rows containing specific value in PySpark dataframe在 PySpark 数据框中删除包含特定值的行
【发布时间】：2019-07-17 12:01:06
【问题描述】：

我有一个 pyspark 数据框，例如：

A    B      C
1    不适用     9
4    2       5
6    4       2
5    1    不适用

我想删除包含值“NA”的行。在这种情况下，第一行和最后一行。如何使用 Python 和 Spark 实现这一点？

根据评论更新：寻找一种解决方案，在许多列中删除具有字符串的行：NA。

【问题讨论】：

NA 是什么意思？这对您来说是否意味着缺失值？还是在您的 DataFrame 中是这样的？在这种情况下，您的 B 列将是一个字符串！请确认。
NA 不是缺失值。这是一个字符串关键字。我想删除所有包含字符串“NA”的行。
此外，NA 也可以出现在另一列中，不一定在 B 列中，因此也应该删除该行。
是的，所以 spark 将它们标记为字符串，因为那里存在“NA”。我想删除“NA”，以便可以将列标记为整数。一种方法是在任何地方用 0 替换“NA”。我无法用正确的语法来实现它。
您应该添加一个最小示例以及到目前为止您已经尝试过的内容。一个简单的过滤器就可以完成这项工作

标签： apache-spark pyspark apache-spark-sql pyspark-sql

【解决方案1】：

只需使用数据框filter 表达式：

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()

【讨论】：

如果有数百列怎么办？
我更新了我的帖子。我认为这是对过滤方法使用字符串表达式的最佳方式。
同意，当然。
我在想的是，如果找到第一个遇到的匹配项，我们如何让它工作并停止？我正在从 Scala 方面寻找它
您想在基于第一行或基于第一列的匹配之后停止？也许有必要提出一个新问题来讨论这个问题。

【解决方案2】：

在 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|
+---+---+---+---+

【讨论】：

【解决方案3】：

如果你想删除行

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))

【讨论】：