PySpark 中的多重过滤

Question

我已经将一个数据集导入到 Juputer notebook/PySpark 中通过 EMR 进行处理，例如：

data sample

我想在使用过滤器功能之前清理数据。这包括：

1. 删除空白行或“0”或不适用的成本或日期。我认为过滤器类似于：.filter(lambda (a,b,c,d): b = ?, c % 1 == c, d = ?)。我不确定如何过滤水果并储存在这里。
2. 删除不正确的值，例如“3”不是水果名称。这对于数字来说很容易（只是数字 % 1 == 数字），但我不确定它会如何过滤掉这些单词。
3. 删除统计异常值的行，即与平均值相差 3 个标准差的行。所以这里的单元格 C4 显然需要删除，但我不确定如何将此逻辑合并到过滤器中。

我需要一次执行一个过滤器，还是有办法一次性过滤数据集（以 lambda 表示法）？

或者，编写一个 Spark SQL 查询是否更容易，而不是在“where”子句中有许多过滤器（但是上面的 #3 仍然难以用 SQL 编写）。

Answer 1

如果您在文档中阅读http://spark.apache.org/docs/2.1.0/api/python/pyspark.sql.html#pyspark.sql.DataFrame.filter，会写到

where() 是 filter() 的别名。

因此，您也可以安全地使用 'filter' 而不是 'where' 来处理多个条件。

编辑： 如果你想过滤很多列的很多条件，我更喜欢这种方法。

from dateutil.parser import parse
import pyspark.sql.functions as F

def is_date(string):
    try: 
       parse(string)
       return True
    except ValueError:
       return False
def check_date(d):
    if is_date(d):
        return d
    else:
        return None

date_udf = F.udf(check_date,StrinType())

def check_fruit(name):
    fruits_list #create a list of fruits(can easily find it on google)
                #difficult filtering words otherwise
                #try checking from what you want, rest will be filtered
    if name in fruits_list:
        return name
    else:
        return None

fruit_udf = F.udf(check_fruit,StringType())

def check_cost(value):
    mean, std #calculcated beforehand
    threshold_upper = mean + (3*std)
    threhold_lower = mean - (3*std)

    if value > threhold_lower and value < threshold_upper:
        return value
    else:
        return None
cost_udf = F.udf(check_cost,StringType())        

#Similarly create store_udf

df = df.select([date_udf(F.col('date')).alias('date'),\
            fruit_udf(F.col('fruit')).alias('fruit'),\
            cost_udf(F.col('cost')).alias('cost'),\
            store_udf(F.col('store')).alias('store')]).dropna()

这将导致所有列一起工作。