使用Python的SequenceMatcher处理7万条推特CSV去重耗时过长求助

嘿，太懂你这种煎熬了——用difflib的SequenceMatcher来处理7万条推特去重，慢到让人崩溃太正常了！毕竟这个方法是逐对做文本相似度比较，时间复杂度基本是O(n²)，数据量上来之后肯定扛不住。下面给你几个从简单到进阶的优化方案，亲测好用：

很多时候，所谓的“重复”其实是完全一模一样的推文（比如重复转发、原封不动的二次发布），先把这些精确重复的内容去掉，能直接砍掉一大半需要处理的数据量，用pandas就能轻松搞定：

import pandas as pd

# 读取CSV文件（注意根据实际情况调整参数，比如编码、分隔符）
df = pd.read_csv('your_tweets.csv')

# 基于Text列去掉完全重复的行，保留第一条出现的记录
df = df.drop_duplicates(subset=['Text'], keep='first')

# 如果想保留最早发布的推文，可以先按时间排序再去重
df = df.sort_values('Creation_Date').drop_duplicates(subset=['Text'], keep='first')

这一步跑完，剩下的数据量会大幅减少，再处理近似重复就轻松多了。

如果你的需求是去重近似相似的推文（比如改写了几个词但核心内容一致），SequenceMatcher的准确率不错但速度拉胯，推荐换成以下两种更高效的算法：

SimHash算法 #

SimHash会把每条文本转换成固定长度的哈希值，通过计算哈希值的汉明距离来判断相似度，速度比逐对比较快几个量级，非常适合大规模文本去重。可以用simhash库实现：

from simhash import Simhash

# 生成每条推文的SimHash值
def get_simhash(text):
    return Simhash(text).value

df['simhash'] = df['Text'].apply(get_simhash)

# 定义汉明距离阈值（比如≤3就认为是近似重复，可根据需求调整）
def is_similar_hash(hash1, hash2, threshold=3):
    return bin(hash1 ^ hash2).count('1') <= threshold

# 遍历过滤重复（如果数据量还是很大，可以用BK树进一步优化查询效率）
seen_hashes = set()
keep_indices = []

for idx, row in df.iterrows():
    current_hash = row['simhash']
    duplicate_found = False
    # 检查已记录的哈希中是否有相似项
    for h in seen_hashes:
        if is_similar_hash(current_hash, h):
            duplicate_found = True
            break
    if not duplicate_found:
        keep_indices.append(idx)
        seen_hashes.add(current_hash)

# 过滤后的结果
df_filtered = df.loc[keep_indices]

MinHash + LSH（局部敏感哈希） #

这套组合拳专门针对大规模数据的近似重复检测：先用MinHash把文本转换成短签名，再用LSH把相似的签名分到同一个“桶”里，只比较同桶内的文本，能大幅减少需要比较的次数。可以用datasketch库实现：

from datasketch import MinHash, MinHashLSH

# 简单的分词函数（如果是英文推文足够用，中文的话可以换成jieba等分词工具）
def tokenize_text(text):
    return text.split()

# 初始化LSH索引，设置相似度阈值（比如0.7，值越高要求越严格）
lsh = MinHashLSH(threshold=0.7, num_perm=128)

minhash_dict = {}
for idx, text in enumerate(df['Text']):
    m = MinHash(num_perm=128)
    # 给每个分词更新MinHash
    for token in tokenize_text(text):
        m.update(token.encode('utf-8'))
    minhash_dict[idx] = m
    lsh.insert(idx, m)

# 找出所有相似的推文组
duplicate_groups = []
seen_ids = set()

for idx in minhash_dict:
    if idx not in seen_ids:
        # 查询与当前推文相似的所有候选
        similar_candidates = lsh.query(minhash_dict[idx])
        duplicate_groups.append(similar_candidates)
        seen_ids.update(similar_candidates)

# 每个相似组只保留第一条（或按Creation_Date保留最早的）
keep_indices = [group[0] for group in duplicate_groups]
df_filtered = df.iloc[keep_indices]

如果你对相似度的要求极高，必须用SequenceMatcher，那可以通过并行计算来加速，把数据拆成多个批次，用多进程同时处理：

from difflib import SequenceMatcher
from concurrent.futures import ProcessPoolExecutor
import pandas as pd

# 判断两条文本是否相似的函数（阈值可调整）
def is_similar(text1, text2, threshold=0.8):
    return SequenceMatcher(None, text1, text2).ratio() >= threshold

# 先做精确去重减少数据量
df = df.drop_duplicates(subset=['Text'])
text_list = df['Text'].tolist()
index_list = df.index.tolist()

# 定义每个进程要处理的分组任务
def process_group(group):
    keep = []
    for i in range(len(group)):
        current_idx, current_text = group[i]
        is_dup = False
        # 和组内之前的文本比较
        for j in range(i):
            if is_similar(current_text, group[j][1]):
                is_dup = True
                break
        if not is_dup:
            keep.append(current_idx)
    return keep

# 把数据分成若干组（比如每组1000条，可根据机器性能调整）
group_size = 1000
groups = [list(zip(index_list[i:i+group_size], text_list[i:i+group_size])) 
          for i in range(0, len(text_list), group_size)]

# 多进程并行处理
with ProcessPoolExecutor() as executor:
    results = executor.map(process_group, groups)

# 合并所有结果
all_keep_indices = []
for res in results:
    all_keep_indices.extend(res)

df_filtered = df.loc[all_keep_indices]

• 如果你的“重复”只定义为完全相同的推文，那直接用pandas的drop_duplicates就够了，完全没必要用近似匹配算法，速度快到飞起。
• 预处理文本：比如把所有文本转小写、去掉标点、删除@用户名和#话题标签（如果这些不影响重复判断），能缩短文本长度，进一步加快匹配速度。
• 超大规模数据的话，可以考虑用数据库工具（比如PostgreSQL的pg_trgm扩展）来做相似查询，性能比纯Python代码更出色。

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