如果文本列包含指定列表中的单词，则过滤 pyspark 数据框

Posted 2023-04-15

【中文标题】如果文本列包含指定列表中的单词，则过滤 pyspark 数据框

【英文标题】：Filtering pyspark dataframe if text column includes words in specified list

【发布时间】：2017-04-25 15:59:55

【问题描述】：

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

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

我尝试了一些方法，但没有成功：

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

返回：TypeError：条件应该是字符串或列

我尝试创建一个函数并使用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))

这会返回一个空数据框，这很好，因为我没有收到任何错误，但显然不是我想要的。

【参考方案1】：

您的 .filter 返回错误，因为它是数据帧上的 sql 过滤器函数（期望 BooleanType() 列）而不是 RDD 上的过滤器函数。如果你想使用RDD，只需添加.rdd：

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

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

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

【参考方案2】：

我认为 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|
+------------------+

【讨论】：

我在我的数据框上尝试了你上面写的 UDF，用我的数据替换了 df.where 和 df.text；但是，我收到错误消息：AttributeError: 'NoneType' object has no attribute 'split'。对于 intersect 函数，从技术上讲，您将列（df.text）作为参数传递，对吗？错误是因为它没有逐行迭代吗？

您的列文本中是否包含空值，这可能解释了 NoneType 错误。您必须在 udf 中将其处理为 ` if row in None return False` 。对于相交行值，即i love Starbucks。如果您将尝试过的内容与示例数据一起发布，这可能会有所帮助。数据图像不是很有帮助。