使用PySpark统计Cloud Storage文本文件中Top10高频单词

你的代码存在几个关键问题：

• 没有利用PySpark的分布式读取能力，反而用google-cloud-storage客户端本地下载文件，失去了Spark的优势
• 自定义words()函数逻辑错误：未定义movies变量、用|分割而非空格、遍历blob对象的方式不正确
• 缺少单词拆分、频率统计、取Top10的核心逻辑

以下是修正后的代码，直接通过PySpark读取GCS文件并完成单词频率统计：

import sys
from pyspark import SparkConf, SparkContext

# 初始化Spark配置与上下文
conf = SparkConf().setMaster("local").setAppName("WordFrequencyTop10")
sc = SparkContext(conf=conf)

# 获取命令行传入的桶名和文件路径
bucket_name = sys.argv[1]
file_path = sys.argv[2]
gcs_full_path = f"gs://{bucket_name}/{file_path}"

# 读取GCS中的文本文件
lines_rdd = sc.textFile(gcs_full_path)

# 处理流程：拆分单词→转为小写→过滤空单词→计数→排序取Top10
word_counts = (
    lines_rdd
    # 拆分每行的单词，空格分隔
    .flatMap(lambda line: line.split(" "))
    # 转为小写，避免大小写导致的重复计数
    .map(lambda word: word.lower())
    # 过滤掉空字符串（比如连续空格产生的）
    .filter(lambda word: word != "")
    # 生成键值对，每个单词计数1
    .map(lambda word: (word, 1))
    # 按单词聚合计数
    .reduceByKey(lambda a, b: a + b)
    # 按计数降序排序，取前10
    .sortBy(lambda x: x[1], ascending=False)
    .take(10)
)

# 输出结果
print("出现频率最高的10个单词：")
for word, count in word_counts:
    print(f"{word}: {count}")

# 停止Spark上下文
sc.stop()

关键说明 #

• 直接读取GCS文件：PySpark支持直接通过gs://路径读取Cloud Storage文件，无需手动下载，充分利用分布式计算能力
• 大小写统一：将所有单词转为小写，避免"Hello"和"hello"被统计为不同单词
• 过滤空单词：处理文本中连续空格产生的空字符串，避免无效计数
• 高效统计：使用flatMap、reduceByKey等Spark原生算子，比本地循环处理更高效，尤其适合大文件

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