本文介绍在 python 中避免因生成全量组合 dataframe 导致内存崩溃的实践方案，重点采用流式写入、条件过滤与多进程并行计算，替代传统 pandas 全量加载方式，适用于 80k 级别词汇的模糊匹配任务。
本文介绍在 python 中避免因生成全量组合 dataframe 导致内存崩溃的实践方案，重点采用流式写入、条件过滤与多进程并行计算，替代传统 pandas 全量加载方式，适用于 80k 级别词汇的模糊匹配任务。
当面对约 80,000 个待清洗词汇（含拼写错误、空格、特殊字符等）并需两两比对相似度时，直接调用 itertools.combinations(my_list, 2) 构建全量组合 DataFrame 将产生约 3.2 × 10⁹ 行（即 32 亿行），远超内存承载能力——即使拥有 32GB RAM，pandas 在内部将迭代器转为列表时仍会触发 MemoryError。根本症结在于：我们并不需要全部组合结果，而仅关注高相似度候选对（如 ratio ≥ 90 或 partial_ratio ≥ 90）。因此，优化核心是「避免构造中间大数据结构，改用流式处理 + 即时过滤 + 持久化落盘」。
✅ 推荐方案一：单进程流式 CSV 写入（简洁可靠，首选）
该方法完全绕过 pandas DataFrame 初始化，使用标准库 csv 模块边计算、边筛选、边写入，内存占用恒定（
import csv
import itertools
from fuzzywuzzy import fuzz # 注意：推荐使用 'thefuzz'（fuzzywuzzy 的维护分支）
MIN_SCORE = 90
# 假设已完成清洗去重（关键预处理！）
my_unique_list = ['apple', 'applee', 'aplee', 'banana', 'banna'] # 示例数据
with open("good_matches.csv", "w", encoding="utf-8", newline="") as f_out:
writer = csv.writer(f_out)
writer.writerow(["name1", "name2", "partial_ratio", "ratio"])
total = len(my_unique_list)
n_combinations = total * (total - 1) // 2
processed = 0
for word1, word2 in itertools.combinations(my_unique_list, 2):
# 即时计算两个指标
partial_ratio = fuzz.partial_ratio(word1, word2)
ratio = fuzz.ratio(word1, word2)
# 仅保留满足阈值的记录（大幅减少 I/O 和存储）
if partial_ratio >= MIN_SCORE or ratio >= MIN_SCORE:
writer.writerow([word1, word2, partial_ratio, ratio])
processed += 1
if processed % 50000 == 0:
print(f"\rProgress: {processed}/{n_combinations} ({processed/n_combinations*100:.1f}%)", end="", flush=True)
print(f"\n✅ Done! Found {processed} total pairs, saved high-score matches to 'good_matches.csv'")
⚠️ 注意事项：
- 务必提前完成文本清洗（如 strip()、正则替换非字母字符、统一小写），否则 fuzz.ratio 效果差且计算冗余；
- fuzzywuzzy 已停止维护，强烈建议迁移到 thefuzz（pip install thefuzz[speedup]），后者支持更快的 C 扩展；
- 阈值 MIN_SCORE 是性能杠杆：设为 95 可使输出量下降 90%+，显著加速。
✅ 推荐方案二：多进程并行加速（适合 CPU 密集型场景）
若单进程耗时过长（如 80k 词需数小时），可引入 multiprocessing 并行化计算。注意：不推荐用 Manager().Queue() 跨进程传大量数据（易阻塞），更优解是各进程独立写入临时文件，最后合并：
import csv
import itertools
import multiprocessing as mp
from pathlib import Path
from thefuzz import fuzz
def process_chunk(chunk_pairs, min_score=90, temp_dir="temp_parts"):
"""每个进程处理一个组合块，写入独立 CSV"""
temp_dir = Path(temp_dir)
temp_dir.mkdir(exist_ok=True)
pid = mp.current_process().pid
temp_file = temp_dir / f"part_{pid}.csv"
with open(temp_file, "w", encoding="utf-8", newline="") as f:
writer = csv.writer(f)
writer.writerow(["name1", "name2", "partial_ratio", "ratio"])
for word1, word2 in chunk_pairs:
p = fuzz.partial_ratio(word1, word2)
r = fuzz.ratio(word1, word2)
if p >= min_score or r >= min_score:
writer.writerow([word1, word2, p, r])
return temp_file
def merge_csv_parts(parts, output_file="good_matches.csv"):
"""合并所有临时文件到最终结果"""
with open(output_file, "w", encoding="utf-8", newline="") as out_f:
writer = csv.writer(out_f)
writer.writerow(["name1", "name2", "partial_ratio", "ratio"])
for part in parts:
with open(part, "r", encoding="utf-8") as f:
reader = csv.reader(f)
next(reader) # skip header
for row in reader:
writer.writerow(row)
part.unlink() # 清理临时文件
if __name__ == "__main__":
my_unique_list = [...] # 你的 80k 清洗后词汇列表
# 划分组合为 chunks（每 chunk ~100k 对，平衡负载）
all_combos = itertools.combinations(my_unique_list, 2)
chunk_size = 100_000
chunks = iter(lambda: list(itertools.islice(all_combos, chunk_size)), [])
# 并行处理
with mp.Pool(processes=mp.cpu_count() - 1) as pool:
temp_files = pool.map(
lambda chunk: process_chunk(chunk, min_score=90),
chunks
)
# 合并结果
merge_csv_parts(temp_files, "good_matches.csv")
print("? Multi-process matching completed!")
? 总结与进阶建议
| 方案 | 适用场景 | 内存占用 | 开发复杂度 | 推荐指数 |
|---|---|---|---|---|
| 流式 CSV（方案一） | 数据量 ≤ 100k，强调稳定性与可读性 | ★☆☆☆☆（极低） | ★☆☆☆☆（简单） | ⭐⭐⭐⭐⭐ |
| 多进程分块（方案二） | 数据量 > 50k，CPU 核心 ≥ 4，追求速度 | ★★☆☆☆（中等） | ⭐⭐⭐☆☆（中） | ⭐⭐⭐⭐☆ |
- 不要做：pd.DataFrame(list(itertools.combinations(...))) —— 这是内存杀手；
- 务必做：清洗前置（去空格、标准化 Unicode、小写）、设置合理相似度阈值；
- 可选增强：对词汇按首字母/长度分桶，仅比对同桶内词对（减少 70%+ 计算量）；
- 生产提示：将 good_matches.csv 加载为 pandas 时，指定 dtype=str 避免类型推断开销。
通过放弃“一次性全量加载”的思维惯性，转向“按需计算 + 流式落盘”，你不仅能解决当前内存问题，更能构建出可扩展、可监控、可部署的文本去重/纠错流水线。
