先说结论线上系统需要处理的“重复文本”真正难点从来不是删除本身而是搞清楚该删哪一层、以什么判据合并、误伤之后怎么回退。拿“你是凑企鹅你是凑企鹅你是凑企鹅”这类字符串来说从内容上看它毫无意义甚至像是机器人刷屏但从数据处理视角看它集合了完全重复、连续片段重复、语义噪声等多个典型问题。用它当例子来走一遍文本去重流程反而比那些规规矩矩的“正经句子”更适合理解边界条件。如果你维护过评论区、论坛、客服工单或者日志采集系统大概率见过同一条消息被重复粘贴 N 次。不同人对此的处理方式也完全不同有人直接调set()有人写正则有人把 Redis 防重 key 一套就以为完事。这些方案都能解决某一种重复但一旦换成“看起来不一样实际上一样”的相似文本或者换成长文本、多行日志、用户生成的极不规范内容就立刻失效。这篇文章不会只给一个set去重的小技巧。我会从“重复文本”的真实类型出发用 Python 写一套能落地的处理链路先做规范化再做精确去重再处理“同一片段连续重复”的情况最后引入 SimHash 做近似文本识别。代码保持独立可跑只依赖 Python 标准库不需要装 sklearn、pandas 这类重型依赖。1. 这篇文章真正要解决的问题很多团队一开始并不会专门设计文本去重功能直到某一天被重复数据打疼了才开始重视。常见的“疼”有三种。第一种是日志存储和检索成本失控。一个应用在异常时打印了一行错误信息循环代码没有做节流或去重一分钟内同类日志可能写几百 MB。到 ELK 或 ClickHouse 里查问题时不仅增加 IO还会让同一条错误在海量结果里反复出现真正需要定位的上下文反而被淹没。这时候需要的是“日志聚合”把相同模式的日志归并成一条并带上时间范围和次数。第二种是用户产生的重复内容泛滥。评论区、留言板、私信里经常有人把同一句话连发三遍意图可能是刷屏、凑字数也可能是操作端逻辑错误导致重复提交。这类文本如果进入数据库会让后来的搜索、统计、内容审核都变得不准确。例如统计某个关键词出现频率时“你是凑企鹅你是凑企鹅你是凑企鹅”会被算成好几次而不是一次。第三种是系统间调用造成的重复数据。下游服务没有做幂等控制触发重试后同一笔数据被写入多条记录。这种重复往往夹杂着时间戳、请求 ID 等动态字段完全相同的字符串去重处理不掉必须做相似度去重或按业务主键去重。换句话说文本重复不是“该不该去掉”的问题而是“应该在入口拦截还是在存储层清理按完全一致去重还是按相似度合并”的问题。只有在动手写代码前把这两件事想清楚才不会出现删完数据才后悔的结果。这问真正要解决的场景包括评论和反馈中的刷屏文本识别、日志系统的重复告警收敛、爬虫或采集任务中重复内容的过滤以及数据库数据清洗时剔除因重试产生的重复记录。建议对文本处理、反垃圾和系统稳定性感兴趣的开发读者仔细看一遍尤其是第五部分的代码和第七部分的排查表基本覆盖了实际项目中几个最容易翻车的点。2. 重复文本的分类完全重复、连续片段重复与近似重复在动手写代码之前先建立一个简单的分类模型。重复文本不是只有“完全一样”这一种形态粗略分可以分成三类。第一类是完全重复。两个字符串在规范化空白字符后完全相同例如你是凑企鹅你是凑企鹅你是凑企鹅 你是凑企鹅你是凑企鹅你是凑企鹅这两行只要去掉换行和空格差异就是同一条文本。最容易处理直接哈希或者放进集合里就能判断。第二类是连续片段重复。字符串本身可能是完整的一句话但其中某个片段在内部重复了多次。比如你是凑企鹅你是凑企鹅你是凑企鹅这个字符串可以看作“你是凑企鹅”这个最小片段重复三次。处理时要做的是把连续重复的片段折叠回一次得到“你是凑企鹅”。这类情况在用户刷屏、程序循环拼接、日志重复写入时非常常见。第三种是近似重复。两条文本并不逐字相同只是大部分信息一致因为时间、编号、标点、语气词或少量无关字符导致内容略有差别。例如系统在 12:00 发生异常请检查服务 系统在12:00发生异常请检查服务两个句子表达几乎一样但直接哈希后会得到完全不同的摘要。这种重复在业务数据中最难处理也是线上问题排查中最容易漏掉的部分。这里引入一个关键词文本指纹。无论是精确去重还是近似去重本质都在比较指纹。精确去重可以直接用 MD5、SHA256 或者 Python 内置的 hash近似去重则需要生成一种能容忍一定差异的指纹指纹很像时认为文本相似。常用的近似文本指纹包括 SimHash、MinHash、LSH 等它们都允许一定误判但计算效率远高于两两计算编辑距离。重复类型典型表现适用处理方式主要风险完全重复两行文本一模一样哈希 集合去重规范化不完整导致漏判连续片段重复“ABABAB”或“ABCABC”正则或周期检测折叠误压缩正常高频词近似重复时间、编号、空格不同内容相同SimHash、编辑距离、LSH阈值不合适造成误判从工程实现的角度看完全重复适合放在最前面做“廉价过滤”因为它速度快、无误差连续片段重复适合做一层“文本清洗”把明显刷屏的文本先折叠近似重复适合放在最后做“兜底”因为它的计算成本最高还需要用业务语料来确定阈值。3. 环境准备与数据约定代码使用 Python 3不锁定具体小版本。只要运行环境是 Python 3.8 或以上代码都可以直接执行。为了减少初学者踩坑这里不引入第三方库只用标准库中的re、hashlib、pathlib、collections和unittest。操作系统方面Windows、macOS、Linux 都兼容。代码中的文件路径做了简单处理如果你在 Linux 上运行可以直接把文件放在当前目录在 Windows 上运行则注意文件编码最好统一使用 UTF-8 保存。本文约定一个最小数据文件作为所有示例的输入。在项目目录下新建demo_comments.txt你是凑企鹅你是凑企鹅你是凑企鹅 这是一条正常评论 这是一条正常评论 系统在 12:00 发生异常请检查服务 系统在12:00发生异常请检查服务 重复内容重复内容重复内容这个文件故意混入了四种情况完全重复行、带内部重复片段的行、看起来相似但空格不同的行还有一个“重复内容重复内容重复内容”的片段重复样例。后续的代码会分别处理这些场景。如果你要在生产环境中处理真实数据请不要直接修改原始文件。更安全的做法是先输出一份清洗后的临时文件确认结果无误后再考虑是否需要覆盖原表或原文件。涉及数据库数据清理时必须先备份并遵循最小影响原则。4. 核心流程拆解一套可落地的文本去重处理链路可以拆成四个步骤。第一步是清洗与规范化。把所有输入内容统一成一种可比较的格式去掉首尾空白、把连续的空白字符压缩成一个空格、统一换行符。对英文文本还要考虑大小写归一化中文文本则要注意全角半角符号比如全角逗号和半角逗号看起来很接近但对哈希算法来说是完全不同的字符。这一层做得越细后面精确去重的效果越好。第二步是精确去重。将清洗后的完整文本计算哈希值并放入集合通过哈希值判断是否已经出现过。若需要保留原始出现顺序可以使用有序字典或setdefault技巧。精确去重速度快但只能解决“完全一样”的问题。第三步是连续片段重复折叠。针对“你是凑企鹅你是凑企鹅你是凑企鹅”这类字符串需要识别出它由某个基本单元重复多次组成。实现时可以利用“字符串乘法”的原理从头开始尝试一个长度递增的周期如果原字符串等于该周期片段重复 N 次就把它替换成单次片段。这一步看起来有点绕但它对刷屏文本的收敛效果非常明显。第四步是近似重复识别。当两段文本因为空格、时间戳、简繁差异导致哈希完全不一致时需要用 SimHash 生成 64 位指纹并比较两个指纹的汉明距离。汉明距离越小说明文本在海明空间里越接近可以近似认为它们是相似文本。实际系统中通常还会建立一个索引避免两两比较带来的 O(n²) 开销。完成这四步之后再根据业务场景决定最终动作。如果是写日志可以只打印聚合后的结果如果是入库评论建议把重复内容折叠掉而不是直接删除原始记录如果是系统间重试产生的重复数据则更应该在入口做幂等控制而不是事后清理。5. 完整示例代码与实现下面这一段是本文的核心我会拆成三个独立的 Python 示例文件。每个示例都可以单独运行最后一个示例把完整流程串起来。5.1 精确去重保留顺序去除重复行# 文件路径exact_dedup.py import re from pathlib import Path def normalize_text(text: str) - str: 清洗单行文本压缩空白、去除首尾空格、统一换行。 text text.replace(\r\n, \n).replace(\r, \n) text re.sub(r\s, , text.strip()) return text def deduplicate_lines(lines): 保留第一次出现顺序的精确去重。 seen set() result [] for line in lines: normalized normalize_text(line) if not normalized: continue if normalized not in seen: seen.add(normalized) result.append(normalized) return result if __name__ __main__: input_path Path(demo_comments.txt) if not input_path.exists(): raise FileNotFoundError(请先创建 demo_comments.txt 文件) raw_lines input_path.read_text(encodingutf-8).splitlines() clean_lines deduplicate_lines(raw_lines) for line in clean_lines: print(line)这个脚本的核心逻辑是seen集合。每读取一行文本先做normalize_text清洗再检查是否已经在集合中。这里使用列表而不是直接输出set是为了保证输出顺序和文件中的首次出现顺序一致。运行方式python exact_dedup.py输入文件中的重复文案会被折叠但“系统在 12:00 发生异常请检查服务”和“系统在12:00发生异常请检查服务”这两行因为在空格数量上不同不会被精确去重处理掉。这正好说明了为什么需要后续的相似度判断。5.2 连续片段重复折叠# 文件路径repeat_compress.py from pathlib import Path def compress_repeated_blocks(text: str) - str: 如果文本由某个子串完整重复 N 次组成则折叠成一次。 if not text: return text length len(text) for period in range(1, length // 2 1): if length % period ! 0: continue block text[:period] if block * (length // period) text: return block return text这段代码利用了一个非常简单的数学性质如果“你是凑企鹅”重复三次后长度为 18那么周期只能是 6、9 等可以整除 18 的长度。代码从周期 1 开始向上试探当发现某个前缀片段重复 N 次能完整还原原字符串时就返回该片段。主入口部分补充为# 文件路径repeat_compress.py续 def main(): lines Path(demo_comments.txt).read_text(encodingutf-8).splitlines() for line in lines: line line.strip() if not line: continue print(compress_repeated_blocks(line)) if __name__ __main__: main()运行后“你是凑企鹅你是凑企鹅你是凑企鹅”会被压缩成“你是凑企鹅”“重复内容重复内容重复内容”会被压缩成“重复内容”。“这是一条正常评论”则保持不变因为没有一个子串能完整覆盖它。这个算法对样本很小的情况效果很好但它也有明显局限只能识别“从头开始就按固定周期重复到结尾”的字符串。如果文本是“你说得对你是凑企鹅你是凑企鹅”核心重复片段出现在句子中间算法会漏掉。生产环境中我们通常配合正则或更复杂的分词手段一起使用。5.3 SimHash 近似去重接下来是近似重复检测。先实现一个基于字符 n-gram 的 SimHash 指纹生成函数。# 文件路径simhash_dedup.py import hashlib import re def build_simhash(text: str, n: int 4) - int: 基于字符 n-gram 构造 64 位 SimHash 指纹。 text re.sub(r\s, , text.strip()) grams [] if len(text) n: grams.append(text) else: for i in range(len(text) - n 1): grams.append(text[i:i n]) vectors [0] * 64 for gram in grams: digest hashlib.md5(gram.encode(utf-8)).digest() value int.from_bytes(digest[:8], big) for bit in range(64): if value (1 bit): vectors[bit] 1 else: vectors[bit] - 1 fingerprint 0 for bit in range(64): if vectors[bit] 0: fingerprint | (1 bit) return fingerprint def hamming_distance(a: int, b: int) - int: 计算两个 64 位整数指纹的汉明距离。 return bin(a ^ b).count(1) def deduplicate_by_simhash(lines, threshold: int 3): 对文本列表做近似去重返回去重后的列表。 fingerprints [] result [] for line in lines: line line.strip() if not line: continue fp build_simhash(line) if all(hamming_distance(fp, old) threshold for old in fingerprints): fingerprints.append(fp) result.append(line) return result这个实现的核心是把每一条文本切成多个长度为 4 的字符片段称为 n-gram。每个片段先 MD5 得到一个整数再把它身上 64 个 bit 的 0/1 状态累加到向量上。最后统计每个 bit 上是 1 多还是 0 多转换成最终的 64 位指纹。这种做法的好处是不需要为中文单独分词连续字符本身就能捕捉局部文字特征。测试入口# 文件路径simhash_dedup.py续 if __name__ __main__: samples [ 系统在 12:00 发生异常请检查服务, 系统在12:00发生异常请检查服务, 这是一条完全不同的评论, ] for item in samples: print(item, 指纹:, hex(build_simhash(item))) print(- * 50) print(近似去重结果:) for item in deduplicate_by_simhash(samples, threshold3): print(item)这里的关键参数是threshold 3。64 位 SimHash 指纹中只要两个指纹不同的位数不超过 3就认为它们是近似重复。生产环境里这个阈值需要根据真实语料调整调小了会漏掉相似文本调大了会把不同主题的文本误合并。近似去重不是银弹。它快是因为把文本压缩成了固定长度的整数指纹但代价是会牺牲可解释性。最终判断时最好保留每个文本的指纹并把重复结果输出到人工复核列表中不要直接自动删除。5.4 完整流程串联把三个模块组合成一个完整流程# 文件路径full_pipeline.py from pathlib import Path from exact_dedup import deduplicate_lines from repeat_compress import compress_repeated_blocks from simhash_dedup import deduplicate_by_simhash def process_text_file(path: str): raw_lines Path(path).read_text(encodingutf-8).splitlines() # 第一步精确去重 exact_unique deduplicate_lines(raw_lines) # 第二步折叠连续重复片段 compressed [compress_repeated_blocks(line) for line in exact_unique] # 第三步再次精确去重因为折叠后可能出现新重复 exact_unique_again deduplicate_lines(compressed) # 第四步近似重复合并 final_lines deduplicate_by_simhash(exact_unique_again, threshold3) return final_lines if __name__ __main__: result process_text_file(demo_comments.txt) for line in result: print(line)这个流程里很容易被忽略的是第三步。连续片段折叠之后可能会出现两条原本不同的文本变成同一条。比如“你好你好你好”和“你好”在折叠前是两个字符串折叠后都是“你好”这时需要再做一次精确去重。6. 运行结果与效果验证完整跑通上面代码后demo_comments.txt的预期输出大致如下但顺序会按代码实际运行结果为准你是凑企鹅 这是一条正常评论 系统在 12:00 发生异常请检查服务最终输出里已经看不到完全重复行也看不到连续重复片段。用户刷屏文本被压缩成有意义的短句相似文本因为 SimHash 识别被判定为同一条。这正好达到前面提出的“先清洗、再精确去重、再片段折叠、最后做近似聚合”的效果。如果你只想快速验证每一段代码是否正确可以换一种方式直接使用unittest写最小断言# 文件路径test_dedup.py import unittest from repeat_compress import compress_repeated_blocks from simhash_dedup import build_simhash, hamming_distance class TestDedupMethods(unittest.TestCase): def test_compress_repeated_blocks(self): self.assertEqual( compress_repeated_blocks(你是凑企鹅 * 3), 你是凑企鹅 ) self.assertEqual( compress_repeated_blocks(这是一条正常评论), 这是一条正常评论 ) def test_simhash_same_text_distance_zero(self): text 系统在12:00发生异常请检查服务 self.assertEqual( hamming_distance(build_simhash(text), build_simhash(text)), 0 ) if __name__ __main__: unittest.main()运行测试命令python -m unittest test_dedup如果所有用例通过就说明最基础的两个算法行为符合预期。建议在你自己的真实数据上继续加大测试集不要只依赖这一条测试。遇到运行失败时先看是不是文件路径或编码问题。Python 在 Windows 上读取 UTF-8 文本偶尔会因默认编码不同出错可以在read_text中显式指定encodingutf-8。另外如果执行python提示找不到命令可以换成python3。7. 常见问题与排查思路文本去重项目里的 bug往往不是算法本身出了大错而是输入数据的复杂程度超过了设计假设。下面这张表列出了经常遇到的问题和排查建议。问题现象可能原因排查方式解决方案两条看起来相同的文本没有被去重空白字符或标点符号不同例如全角逗号和半角逗号打印文本的repr()查看不可见符号扩展normalize_text统一全角半角和空白使用set后输出顺序随机set本身是无序结构检查代码是否直接遍历集合改为dict.fromkeys或列表 集合连续重复片段压缩误伤高频词“哈哈”这类本身有意义的短语被压缩加入人工样本测试设置最小重复次数阈值只处理连续出现 2 次以上的场景SimHash 误判率较高n-gram 长度或汉明距离阈值不合适用小批量人工标注数据验证调整 n 和 threshold必要时换用 MinHash日志量很大去重任务跑得很慢对所有文本做两两 SimHash 比较打印耗时量化 O(n²) 阶段对指纹建立分桶索引或使用布隆过滤器粗筛数据清理后想恢复原始记录没有在清理前做备份检查是否有备份或快照生产环境禁止直接删除源数据应输出到新表再原子切换很多人第一次做文本去重时会以为只要“把重复的删掉”就够了。真正动手后才发现去重只是信息提取的前置步骤。正确做法是保留一份原始数据把去重后的数据用于统计、搜索或告警原始数据即使暂时用不上也最好不要当下就删尤其是业务流程尚未完全稳定时。8. 最佳实践与工程建议经过几轮迭代之后我建议把文本去重方案放到系统的数据入口层而不是等到数据已经落库后再做全量清洗。入口拦截的收益远大于事后清洗因为事后清洗要面对百倍甚至千倍的数据量还容易影响线上查询。以下几条是实际项目中比较通用的建议。首先命名和职责要清晰。不要写一个叫clean_text的大函数把所有逻辑都塞进去。更合理的拆分是normalize_text、hash_dedup、compress_repeated_blocks、simhash_dedup每一层只做一件事测试也更容易写。后续如果调整某个算法不至于动到整条链路。其次明确幂等与去重的边界。文本去重不是系统幂等的替代品。如果你的接口收到同一笔订单请求并发生重试正确方案是在接口层使用业务主键做幂等判断例如订单号、请求 ID而不是拿整个响应文本去重。拿文本去重处理这类问题很可能因为内容包含动态时间戳而导致完全失效。第三对不同来源的数据做分级处理。用户评论和系统日志的处理策略不应该一样用户评论重复一次可能是误操作重复三次通常是刷屏日志中同一条错误重复出现则不一定需要去重可能反而说明循环代码有问题。日志场景下更推荐在上游先做聚合而不是在下游反复清理。如果需要在日志里快速统计哪一类错误出现最多可以先使用 shell 命令做一次粗糙统计再决定要不要写完整程序grep ERROR app.log | sort | uniq -c | sort -nr | head -20这行命令的含义是过滤出包含ERROR的日志行排序后统计每一行出现的次数再按次数倒序排列只看次数最高的 20 条。它适合临时快速定位热点错误但它只是字符串级统计对于“内容相似但时间戳不同”的日志无能为力。第四给结果留出审计路径。去重系统一旦自动化运行就可能出现“程序把不该合并的文本合并了”的情况。遇到这种场景工程师最怕的不是误判而是查不到原因。所以输出结果时最好一并输出指纹和被拒文本的原因数据库清洗时更要保留操作日志。第五评估成本和收益。近似去重算法不是免费的。如果不缺存储、数据量也只有几万条直接两两计算编辑距离都比实现一套 SimHash 简单直接。选择算法之前先统计数据规模、重复率、重复类型之后再决定技术方案。9. 总结与后续学习方向文本去重这个主题听起来很小实际拆开后涉及哈希、字符串算法、相似度计算、工程分层等多个知识点。这篇文章用“你是凑企鹅你是凑企鹅你是凑企鹅”这类重复文本作为切入点梳理了完全重复、连续片段重复和近似重复三种类型并给出了完整的 Python 实现链路。代码可以直接复制保存为三个脚本在本地用一个很小的文本文件跑通流程再逐步替换成自己的业务数据。看完这篇文章你可以先做一个小练习把自己项目里的日志或评论文件抽一千条先用精确去重统计重复率再用 SimHash 抽样看有多少“看起来不同、实际含义接近”的文本。如果不做这个抽样你可能永远意识不到自己系统里有多少近似重复数据。下一步可以继续研究的方向包括对超大文本集合使用 MinHash LSH 加速检索、用 Redis 的布隆过滤器做重复请求的快速拦截、在消息队列消费者中实现消息幂等。文本去重的核心思想一旦掌握切到任何场景都只是换一层外壳。这里也建议把上面的示例和排查表收藏起来等真正需要做内容清洗或日志聚合时再对照着手会更快。