如何在PySpark中拆分JSON推文文本并提取含#的词汇

实现步骤 #

1. 导入内置函数库 #

首先导入PySpark SQL的函数模块，后续的文本处理操作都依赖这些内置函数：

from pyspark.sql import functions as F

2. 生成全词汇数组 #

先清理文本首尾的多余空格，再按任意数量的空白字符拆分文本，得到包含所有词汇的数组列：

tweets = tweets.withColumn("all_words", F.split(F.trim(F.col("text")), "\\s+"))

• trim(F.col("text"))：去除文本前后的空格、制表符等空白字符
• split(..., "\\s+")：用正则匹配连续空白字符作为分隔符，避免因多个空格导致的空元素

3. 生成含#的词汇数组 #

基于全词汇数组，筛选出以#开头的元素，得到话题标签词汇数组：

tweets = tweets.withColumn("hashtag_words", F.filter(F.col("all_words"), lambda x: x.startswith("#")))

如果你的PySpark版本低于3.0（不支持lambda参数），可以用正则匹配替代：

tweets = tweets.withColumn("hashtag_words", F.filter(F.col("all_words"), x=x.rlike("^#")))

4. 查看处理结果 #

执行以下命令查看新增的数组列：

tweets.select("text", "all_words", "hashtag_words").show(truncate=False)

完整代码整合 #

import findspark
findspark.init()

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession \
        .builder \
        .appName("Jupyter Spark shell") \
        .getOrCreate()

folder = 'tweet-id-text-345'
# 修正原代码的参数拼写错误：delimeter → delimiter
tweets = spark.read.format("json").option("delimiter", "\t").load(folder)

# 生成全词汇数组
tweets = tweets.withColumn("all_words", F.split(F.trim(F.col("text")), "\\s+"))
# 生成话题标签数组
tweets = tweets.withColumn("hashtag_words", F.filter(F.col("all_words"), lambda x: x.startswith("#")))

tweets.count()
tweets.select("text", "all_words", "hashtag_words").show(truncate=False)

注意：原代码中option("delimeter", "\t")存在拼写错误，正确参数名为delimiter，已在代码中修正。

内容的提问来源于stack exchange，提问作者Roald Andre Kvarv