如何用Python从数百万份文档中查找最常见的长句/短语（至少5词）

前置依赖安装 #

你需要提前安装所需第三方库：

pip install nltk tqdm
# 如果处理中文文档，额外安装jieba
pip install jieba

首次运行需要下载nltk的分句分词数据包：

import nltk
nltk.download('punkt')

核心实现代码 #

1. 工具函数定义 #

import os
from collections import Counter
from nltk.tokenize import sent_tokenize, word_tokenize
from tqdm import tqdm

# 中文处理请取消下面两行注释，替换分词逻辑
# import jieba
# def word_tokenize(text):
#     return jieba.lcut(text)

def extract_valid_phrases(text, min_length=5):
    """从文本中提取长度≥指定单词数的短语/句子"""
    phrases = []
    # 拆分为独立句子
    sentences = sent_tokenize(text)
    for sent in sentences:
        # 分词并过滤纯标点符号token
        words = [w for w in word_tokenize(sent) if w.strip() and not w in ',.!?"\'[]{}();:']
        if len(words) < min_length:
            continue
        # 统计完整句子用下面这行
        phrases.append(tuple(words))
        # 如需统计所有连续≥5词的短语，取消下面两行注释
        # from nltk.util import ngrams
        # phrases.extend([ngram for n in range(min_length, len(words)+1) for ngram in ngrams(words, n)])
    return phrases

2. 批量统计逻辑 #

采用流式分批处理，避免一次性加载所有文档导致内存溢出：

def batch_count_phrases(doc_root_path, min_length=5, batch_size=10000):
    total_counter = Counter()
    # 遍历所有纯文本文件
    all_txt_files = [os.path.join(doc_root_path, f) for f in os.listdir(doc_root_path) if f.lower().endswith('.txt')]
    # 分批处理
    for batch_idx in tqdm(range(0, len(all_txt_files), batch_size)):
        batch_files = all_txt_files[batch_idx: batch_idx + batch_size]
        batch_counter = Counter()
        for file_path in batch_files:
            try:
                with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
                    content = f.read()
                phrases = extract_valid_phrases(content, min_length)
                batch_counter.update(phrases)
            except:
                # 跳过损坏、无法读取的文档
                continue
        # 合并批次统计结果，可选过滤掉批次内仅出现1次的低频短语减少内存占用
        total_counter.update({k:v for k,v in batch_counter.items() if v >= 2})
    return total_counter

3. 结果输出 #

if __name__ == "__main__":
    # 替换为你的文档所在目录
    DOC_DIR = "./your_documents_path"
    # 统计所有短语
    result_counter = batch_count_phrases(DOC_DIR, min_length=5)
    # 输出出现频率最高的前100个结果
    top_100 = result_counter.most_common(100)
    # 保存结果到文件
    with open("./high_freq_phrases.txt", "w", encoding="utf-8") as f:
        for phrase_tuple, count in top_100:
            phrase_str = " ".join(phrase_tuple)
            f.write(f"{phrase_str}\t{count}\n")
            print(f"短语：{phrase_str}，出现次数：{count}")

大规模场景优化 #

如果文档总大小超过100G，可做如下调整提升效率：

• 用multiprocessing多进程并行处理不同批次的文档，最后再聚合各进程的统计结果
• 每批处理完后仅保留出现次数≥2的短语，大幅降低内存占用
• 超大规模场景可改用Spark分布式框架实现，核心逻辑和上述方案一致，仅需将统计逻辑迁移到RDD/DataFrame算子中即可

注意：如果需要统计的是连续5个汉字而非5个词，调整分词逻辑和长度判断规则即可。

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