如果文本列包含指定lis中的单词，则筛选pyspark数据帧

Question

我在这里看到了一些与我的问题类似的问题，但是当我尝试一些被接受的答案时，我的代码仍然有错误。我有一个包含三列的数据框架——创建的、文本和单词（这只是文本的标记化版本）。见下文：

现在，我有一个公司列表['Starbucks', 'Nvidia', 'IBM', 'Dell']，我只想保留文本中包含上面这些词的行。

我试过几次，但没有成功：

small_DF.filter(lambda x: any(word in x.text for word in test_list))

返回：TypeError:条件应为string或Column

我尝试创建一个函数并使用foreach()：

def filters(line):
   return(any(word in line for word in test_list))
df = df.foreach(filters)

把df变成“Nonetype”

我最后一次尝试：

df = df.filter((col("text").isin(test_list))

这将返回一个空数据帧，这很好，因为我没有得到错误，但显然不是我想要的。

Answer 1

我认为filter不起作用，因为它需要lambda函数的布尔输出，而isin只与列进行比较。您正在尝试将单词列表与单词列表进行比较。我试着给你一些指导-

# prepare some test data ==> 

words = [x.lower() for x in ['starbucks', 'Nvidia', 'IBM', 'Dell']]
data = [['i love Starbucks'],['dell laptops rocks'],['help me I am stuck!']]
df = spark.createDataFrame(data).toDF('text')


from pyspark.sql.types import *

def intersect(row):
    # convert each word in lowecase
    row = [x.lower() for x in row.split()]
    return True if set(row).intersection(set(words)) else False


filterUDF = udf(intersect,BooleanType())
df.where(filterUDF(df.text)).show()

输出：

+------------------+
|              text|
+------------------+
|  i love Starbucks|
|dell laptops rocks|
+------------------+

Answer 2

您的.filter返回一个错误，因为它是dataframes上的sql filter函数（需要BooleanType()列），而不是rdd上的filter函数。如果要使用RDD，只需添加.rdd：

small_DF.rdd.filter(lambda x: any(word in x.text for word in test_list))

您不必使用UDF，您可以在pyspark中使用正则表达式，列上有.rlike：

from pyspark.sql import HiveContext
hc = HiveContext(sc)
import pyspark.sql.functions as psf

words = [x.lower() for x in ['starbucks', 'Nvidia', 'IBM', 'Dell']]
data = [['i love Starbucks'],['dell laptops rocks'],['help me I am stuck!']]
df = hc.createDataFrame(data).toDF('text')
df.filter(psf.lower(df.text).rlike('|'.join(words)))