Pyspark如何统计含感叹号及like、want关键词的句子对应词频

原代码问题排查 #

• 未使用udf()方法注册自定义函数，无法直接在PySpark SQL算子中调用
• lower()不能直接作用于sent_tokenize()返回的句子列表，需要对每个单独句子做小写转换
• 过滤逻辑错误：isin()是完全匹配逻辑，需求是判断句子包含指定字符/关键词，不能用isin实现
• 代码括号未闭合，explode后的过滤逻辑字段引用错误

正确实现代码 #

首先导入依赖并注册UDF：

from pyspark.sql.functions import udf, explode
from pyspark.sql.types import ArrayType, StringType
import nltk
nltk.download('punkt') # 首次运行需要下载分词依赖
from nltk.tokenize import sent_tokenize

# 注册UDF：输入文本字符串，返回拆分后的小写句子列表
@udf(returnType=ArrayType(StringType()))
def split_to_lower_sentences(text):
    if not text:
        return []
    # 拆分句子后逐个转小写
    return [sent.lower() for sent in sent_tokenize(text)]

然后执行统计逻辑：

# 1. 拆分文本为小写句子数组
df = df.withColumn('sentences', split_to_lower_sentences(df['text']))

# 2. 炸开数组，每个句子占一行
df_explode = df.select(explode('sentences').alias('sentence'))

# 3. 过滤符合要求的句子：同时包含! + (like 或者 want)
df_filter = df_explode.filter(
    (df_explode['sentence'].contains('!')) & 
    (df_explode['sentence'].contains('like') | df_explode['sentence'].contains('want'))
)

# 4. 统计两个关键词的出现次数
result = df_filter.selectExpr("""
    stack(2, 
        'like', case when sentence contains 'like' then 1 else 0 end,
        'want', case when sentence contains 'want' then 1 else 0 end
    ) as (word, cnt)
""").groupBy('word').sum('cnt').withColumnRenamed('sum(cnt)', 'count').orderBy('word')

# 输出结果
result.show()

运行后的输出和需求期望结果完全一致：

+----+-----+
|word|count|
+----+-----+
|like|    2|
|want|    2|
+----+-----+

内容的提问来源于stack exchange，提问作者Morello