PySpark作业性能优化问询：短语计数转词频统计优化方案

针对(短语\t计数)格式RDD的唯一词数、TopN/BottomN词频统计场景，当前实现因数据无意义展开导致性能低下，以下是针对性优化方案：

当前代码的核心问题 #

• 数据冗余膨胀：通过range(int(line.split("\t")[1]))重复展开短语，大数据场景下会生成海量冗余数据，严重消耗内存、磁盘IO和网络带宽
• Driver端过载：countByValue()将全量词频结果拉取到Driver端，数据量较大时易引发内存溢出；后续本地排序完全依赖Driver算力，效率极低
• 冗余计算：同一行数据重复调用line.split("\t")，增加不必要的计算开销

优化思路：基于reduceByKey的分布式累加 #

核心逻辑是直接传递计数进行分布式累加，避免展开短语的无意义操作，全程利用Spark分布式计算能力：

1. 拆分每条记录为(短语, 计数)，将短语拆分为单词后，每个单词关联原计数（短语出现count次，等价于每个单词的计数增加count）
2. 用reduceByKey分布式累加每个单词的总出现次数
3. 利用Spark原生分布式API完成唯一词数统计、TopN/BottomN提取，避免全量数据拉到Driver

优化后的代码 #

def top_bottom_words(resilient_dd, n):
    """
    输出唯一单词总数、出现次数最多的n个单词、出现次数最少的n个单词及对应计数

    参数:
    resilient_dd: 格式为(短语\t计数)的Resilient Distributed Dataset
    n: 需要输出的TopN和BottomN的数量

    返回:
    total: 词汇表中唯一单词的总数
    n_most_frequent: 出现次数最多的n个单词及对应计数
    n_least_frequent: 出现次数最少的n个单词及对应计数
    """
    # 1. 拆分记录并转换为(单词, 计数)键值对
    word_count_pairs = resilient_dd.flatMap(lambda line: 
        # 一次拆分获取短语和计数，避免重复计算；处理单词小写统一格式
        [(word.lower(), int(count)) for word in phrase.split(" ")]
        for phrase, count in [line.split("\t", 1)]  # split("\t",1)避免短语中含制表符的情况
    )

    # 2. 分布式累加每个单词的总计数
    word_counts = word_count_pairs.reduceByKey(lambda a, b: a + b)

    # 3. 统计唯一单词总数
    total = word_counts.count()

    # 4. 获取TopN单词（分布式计算，无需拉全量数据到Driver）
    n_most_frequent = word_counts.top(n, key=lambda x: x[1])

    # 5. 获取BottomN单词（用takeOrdered实现分布式排序取前n）
    n_least_frequent = word_counts.takeOrdered(n, key=lambda x: x[1])

    return total, n_most_frequent, n_least_frequent

关键优化点说明 #

• 避免数据膨胀：直接将每个单词与原短语的计数关联，不再生成重复的单词实例，数据量仅与短语中的单词总数成正比，而非总计数之和
• 分布式计算最大化：reduceByKey在Executor端完成局部累加，仅传递最终结果；top和takeOrdered通过分布式排序获取结果，无需拉取全量数据到Driver
• 减少冗余操作：用split("\t",1)一次拆分短语和计数，避免重复调用split；同时统一处理单词小写，减少重复计算
• 移除不必要的持久化：原代码中persist的RDD仅使用一次，persist反而增加内存开销，优化后无需额外持久化

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