PySpark standalone集群3个Worker执行词频统计出现单词重复问题咨询

• 数据源读取使用了非共享的本地路径：你使用/home/ubuntu/sparkscripts/files/这个本地路径作为输入，3个Worker节点的相同路径下都存放了完全一致的源文件，Spark默认调度Worker优先读取本地文件，导致整个数据集被重复加载了3次。
• 未触发全局shuffle聚合：reduceByKey需要跨节点shuffle才能实现全局的计数聚合，你当前的集群任务调度将3份重复数据分别放在3个Worker节点独立计算，没有触发跨节点的shuffle逻辑，每个Worker仅生成自身读取数据的统计结果。
• coalesce(1)特性导致结果直接拼接：coalesce默认不触发shuffle，仅将多个分区的内容直接合并为单个文件，不会做二次聚合，所以最终输出里每个单词会对应3个Worker的3条独立统计值，和你观察到的现象完全匹配。

• 统一数据源存储
  把源文件上传到所有Worker可统一访问的分布式存储（如HDFS、集群共享NAS），读取时使用对应的分布式路径，例如HDFS路径格式为hdfs://<namenode地址>:<端口>/<文件路径>，确保所有节点读取同一份源数据，不会重复加载。如果暂时不用分布式存储，提交任务时添加--files <源文件路径>参数，把源文件同步到所有Worker的工作目录，避免各节点读取本地冗余文件。
• 修正读取逻辑&清理历史文件
  每次运行任务前先删除输出目录，避免旧输出被误读入计算。不要直接读取包含输出子目录的父目录作为输入，可指定读取的文件后缀过滤：

# 仅读取txt后缀的源文件，避免读取output子目录的内容
words = sc.textFile("/home/ubuntu/sparkscripts/files/*.txt").flatMap(lambda line: line.split(" "))

• 确保全局聚合生效
  如果确认shuffle配置异常导致聚合不生效，可以把coalesce(1)替换为repartition(1)，repartition会强制触发shuffle，完成全局聚合后再合并为单个分区输出：

wordCounts.repartition(1).saveAsTextFile("/home/ubuntu/sparkscripts/files/output/")

也可以在输出前再做一次全局聚合，确保每个单词仅保留一条统计值：

wordCounts = wordCounts.reduceByKey(lambda a, b: a + b)
wordCounts.coalesce(1).saveAsTextFile("/home/ubuntu/sparkscripts/files/output/")

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