先梳理下你的核心实现逻辑：

• 多线程模拟Map阶段：每个线程初始化一个包含所有语料库词汇的字典（初始值为0），遍历分配到的文档并递增对应词汇的计数，完成后将结果保存为JSON文件
• Reduce阶段汇总：通过merge_counts方法加载第一个JSON文件作为初始计数，再遍历剩余所有JSON文件逐个累加对应词汇的计数

现有方案可能存在的问题 #

1. 内存占用过高：每个线程都持有全量词汇的字典，如果语料库词汇量达到几十万甚至百万级，多线程同时运行时会占用大量内存，线程数越多内存压力越大
2. IO与序列化开销大：全量字典写入JSON文件的体积大，后续汇总时逐个加载反序列化的耗时也会很高
3. 冗余存储浪费资源：单个线程处理的文档中，大部分词汇可能根本没出现，但依然要在字典中占据存储空间

优化建议 #

1. 调整Map阶段的输出策略

让每个线程只保存实际出现过的词汇及其计数，而非全量词汇字典，能大幅降低内存占用和文件体积：

def map_worker(self, docs):
    count_dict = {}
    for doc in docs:
        for word in doc.split():
            count_dict[word] = count_dict.get(word, 0) + 1
    # 仅保存出现过的词汇到文件
    output_path = os.path.join(self.get_current_job_directory(), f"thread_{threading.get_ident()}.json")
    with open(output_path, 'w') as f:
        json.dump(count_dict, f)

2. 简化Reduce阶段的汇总逻辑

不需要依赖第一个文件作为初始值，直接初始化空字典遍历所有文件累加即可，逻辑更简洁：

def merge_counts(self):
    job_dir = self.get_current_job_directory()
    count_files = [f for f in os.listdir(job_dir) if f.endswith('.json')]
    total_counts = {}
    
    for file_path in count_files:
        full_path = os.path.join(job_dir, file_path)
        with open(full_path, 'r') as f:
            thread_counts = json.load(f)
        # 累加每个词汇的计数
        for word, cnt in thread_counts.items():
            total_counts[word] = total_counts.get(word, 0) + cnt
    return total_counts

3. 用更高效的存储格式替代JSON

JSON的序列化/反序列化效率不算最优，对于大量小文件场景，可以换成Python原生的pickle（序列化速度更快）或者msgpack（体积更小且跨语言）：

# 用pickle保存线程计数结果
import pickle
def map_worker(self, docs):
    # ... 计数逻辑同上 ...
    with open(output_path, 'wb') as f:
        pickle.dump(count_dict, f)

# 汇总时读取pickle文件
def merge_counts(self):
    # ... 获取文件列表 ...
    for file_path in count_files:
        with open(full_path, 'rb') as f:
            thread_counts = pickle.load(f)
        # ... 累加逻辑同上 ...

4. 内存敏感场景的进阶方案

如果语料库词汇量极大，即使优化后内存依然紧张，可以用磁盘存储替代内存累加，比如用SQLite数据库逐步更新计数：

import sqlite3

def merge_counts(self):
    conn = sqlite3.connect(os.path.join(self.get_current_job_directory(), 'word_counts.db'))
    cursor = conn.cursor()
    cursor.execute('CREATE TABLE IF NOT EXISTS counts (word TEXT PRIMARY KEY, count INTEGER DEFAULT 0)')
    
    job_dir = self.get_current_job_directory()
    count_files = [f for f in os.listdir(job_dir) if f.endswith('.json')]
    
    for file_path in count_files:
        with open(os.path.join(job_dir, file_path), 'r') as f:
            thread_counts = json.load(f)
        # 批量插入或更新计数
        for word, cnt in thread_counts.items():
            cursor.execute(
                'INSERT INTO counts (word, count) VALUES (?, ?) ON CONFLICT(word) DO UPDATE SET count = count + ?',
                (word, cnt, cnt)
            )
    conn.commit()
    
    # 导出最终结果
    cursor.execute('SELECT word, count FROM counts')
    total_counts = {row[0]: row[1] for row in cursor.fetchall()}
    conn.close()
    return total_counts

总结 #

你的分治思路是对的，通过调整Map阶段的输出内容、优化Reduce的累加逻辑以及选择更高效的存储方式，可以大幅提升整个流程的内存利用率和运行速度。

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