使用来自另一个 DataFrame 的关键字过滤 Spark DataFrame

Question

我有一个大型新闻文章数据集加载到 PySpark DataFrame 中。我有兴趣将该 DataFrame 过滤到正文中包含某些感兴趣的单词的文章集。目前关键字列表很小，但我还是想将它们存储在 DataFrame 中，因为该列表将来可能会扩展。考虑以下小例子：

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

article_df = [{'source': 'a', 'body': 'Seattle is in Washington.'},
                {'source': 'b', 'body': 'Los Angeles is in California'},
                {'source': 'a', 'body': 'Banana is a fruit'}]
article_data = spark.createDataFrame(article_data)

keyword_data = [{'city': 'Seattle', 'state': 'Washington'},
                {'city': 'Los Angeles', 'state': 'California'}]
keyword_df = spark.createDataFrame(keyword_data)

这为我们提供了以下 DataFrame：

+--------------------+------+
|                body|source|
+--------------------+------+
|Seattle is in Was...|     a|
|Los Angeles is in...|     b|
|   Banana is a fruit|     a|
+--------------------+------+

+-----------+----------+
|       city|     state|
+-----------+----------+
|    Seattle|Washington|
|Los Angeles|California|
+-----------+----------+

作为第一遍，我想过滤掉article_df，使其仅包含body 字符串包含keyword_df['city'] 中的任何字符串的文章。我还想将其过滤到包含来自keyword_df['city'] 的字符串和keyword_df['state'] 中的相应条目（同一行）的文章。我怎样才能做到这一点？

我已经设法使用手动定义的关键字列表来做到这一点：

from pyspark.sql.functions import udf
from pyspark.sql.types import BooleanType
def city_filter(x):
    cities = ['Seattle', 'Los Angeles']
    x = x.lower()
    return any(s.lower() in x for s in cities)
filterUDF = udf(city_filter, BooleanType())

然后article_df.filter(filterUDF(article_df.body)).show() 给出想要的结果：

+--------------------+------+
|                body|source|
+--------------------+------+
|Seattle is in Was...|     a|
|Los Angeles is in...|     b|
+--------------------+------+

如何在无需手动定义关键字列表（或关键字对元组）的情况下实现此过滤器？我是否需要为此使用 UDF？

Answer 1

您可以使用 leftsemi 与自定义表达式连接来实现它，例如：

body_contains_city = expr('body like concat("%", city, "%")')
article_df.join(keyword_df, body_contains_city, 'leftsemi').show()