在采集用户留言、论坛帖子或接口返回文本时我见过不少类似这样的样本你是凑企鹅你是凑企鹅你是凑企鹅。第一次看像乱码复制到编辑器里搜索才发现是同一句话被连续拼了三次。它不是人写的正常表达也不是普通口语而是一段带有明显重复噪音的数据。如果直接把这类文本送去分词、聚类或做关键词统计轻则污染词频重则让相似度模型把两条完全无关的文本误判成同一类。更隐蔽的一点是这类重复文本很难用“完整行去重”一次清掉因为它在行内已经重复了而字典、集合类工具只能处理“行与行之间的完全重复”。这篇文章围绕这条脏数据展开讲清楚重复文本为什么会出现、怎么识别、怎么清洗、怎么验证以及过程中有哪些容易踩的坑。我会给出一个可运行的最小 Python 清洗项目同时介绍 Linux 下的sort、uniq、awk、perl组合用法。读完以后你可以把这套方法用到日志清洗、评论去重、舆情数据处理和 NLP 训练集预处理中。1. 先弄清楚重复拼接文本算哪种脏数据1.1 无语义重复与自然重复并不是一回事你是凑企鹅你是凑企鹅你是凑企鹅是一个无语义重复的典型样本。整段文字没有新增信息只有同一句话的机械循环。人一眼能看出来的原因是它完整复现了同一个子串并且重复边界非常整齐。自然语言中的“重复”则完全不同。比如“我说我说我说你还不信”虽然也出现了连续的“我说”但它在表达情绪、语气和停顿并不是单纯的数据冗余。如果清洗逻辑把所有连续出现的短语都压缩成一个就会把这类正常表达也破坏掉。因此在设计清洗规则前必须先区分三种情况类型示例是否该清理处理难点整行完全重复同一行内容出现 100 次应该清理简单用去重集合即可行内连续重复你是凑企鹅你是凑企鹅你是凑企鹅通常应该压缩为单个短语需要识别重复边界语义级重复“这台手机很好用”和“这手机使用体验不错”需要结合业务判断依赖语义模型或人工规则语气重复“好好好我知道了”不能一律压缩需要看重复单元长度和上下文一般做文本预处理时首先处理的是前两类。因为它们可以通过规则和字符串算法稳定识别误判风险较低。第三类属于语义去重通常放在检索、聚类或模型训练阶段处理不能直接使用简单的文本替换。1.2 这类脏样本通常从哪里来我在实际日志和数据仓库任务里见过的重复拼接文本来源主要有四类。第一类是前端或客户端在采集时做了循环拼接。比如一个字段值被循环写入写入次数来自错误的上游参数最后形成一整段无意义的字符串。第二类是爬虫或数据同步任务在失败重试时把上一次的部分结果继续追加到新结果后面。如果两条记录本身内容相似拼接后的文本会出现前后重复段。第三类是日志框架的堆栈字段被重复输出。比如某些监控系统把异常信息聚合到一起每聚合一次就把原文本追加一遍最终在消息队列里留下大量重复行。第四类是用户刷屏或恶意灌水。匿名评论、问答平台经常有人把一句话连续发布多次或者通过脚本生成大量近似文本目的是提高曝光量或干扰自动审核。这些来源意味着你很难用一句“数据采集方已经做过清洗”来替自己开脱数据处理链路的每一层都可能有责任做一次防御性清洗。1.3 不处理会对后续环节造成什么污染如果带着这类文本进入下游影响会逐层放大。分词阶段像 jieba 这类基于词频和前缀的模型会把重复片段识别成更长的未知词导致词边界错误。比如你是凑企鹅你是凑企鹅你是凑企鹅按字符被切成若干个不稳定片段词频统计结果会明显偏向重复词。关键词提取阶段TF-IDF 会把出现次数多的组成部分判断为关键词。一段重复文本本身占了几百字分词后的字符数量虚高会让“凑”、“企鹅”这类片段在关键词权重里异常靠前。文本聚类和相似度检索阶段重复文本会造成假相似。如果两条消息都包含同一段被重复拼接的内容即使主体信息完全不同字面特征也会让它们表现得很相似从而被错误归入同一个类。模型训练阶段重复样本会放慢收敛速度并造成训练集和验证集之间泄漏。若同一文本以不同重复形式同时出现在训练集、测试集评估指标就会虚高线上表现却明显偏离。所以清洗重复文本不是洁癖而是数据质量建设的一环。处理优先级可以放在格式归一化和缺失值填充之后但要放在特征提取和建模之前。2. 环境准备搭建一个最小可运行的清洗实验目录2.1 建立目录并准备样例数据这一节我们先在本地跑通一个最小闭环。实验环境只需要 Python 3.8 以上版本不需要安装第三方依赖。目录结构建议如下text_cleaner/ ├── data/ │ ├── raw.txt │ └── normalized.txt ├── scripts/ │ ├── inspect.py │ └── clean.py └── output/ ├── dedup_lines.txt └── repeat_compressed.txt先创建目录并准备一个包含重复噪音的文本文件。raw.txt里我故意混入几种类型的脏数据方便观察不同规则的清洗效果。mkdir -p text_cleaner/data text_cleaner/scripts text_cleaner/output样本内容如下你是凑企鹅你是凑企鹅你是凑企鹅 今天天气不错适合跑步 今天天气不错适合跑步 你是凑企鹅你是凑企鹅你是凑企鹅 这个方案需要再评审一次 这个方案需要再评审一次 这个方案需要再评审一次 收到明天下午见把上面内容保存为text_cleaner/data/raw.txt。这里有一行是长度为 6 的重复短语拼接有三行是整行重复还有一行是正常文本。这样构造可以同时验证“行间去重”和“行内压缩”两个逻辑。2.2 先做数据体检不要急着写清洗函数写清洗脚本之前先统计原始文件的行数、空行数、重复占比和平均长度。这样能知道当前数据的脏程度也能在清洗后对比处理效果。下面是一个小脚本scripts/inspect.py只使用标准库。from collections import Counter from pathlib import Path def inspect_text(path: str) - None: text Path(path).read_text(encodingutf-8) lines text.splitlines() non_empty_lines [line.strip() for line in lines if line.strip()] line_counter Counter(non_empty_lines) duplicate_lines { line: count for line, count in line_counter.items() if count 1 } total_chars sum(len(line) for line in non_empty_lines) print(f文件路径: {path}) print(f非空行数: {len(non_empty_lines)}) print(f总字符数: {total_chars}) print(f平均行长度: {total_chars / max(1, len(non_empty_lines)):.2f}) print(重复行统计:) for line, count in duplicate_lines.items(): print(f {count} 次 - {line}) if __name__ __main__: inspect_text(data/raw.txt)运行命令cd text_cleaner python scripts/inspect.py预期的输出会显示非空行数为 8有两行内容重复出现 2 次有一行内容重复出现 3 次。这段输出先帮你确认基础重复规模后续清洗是“减少多少重复行”就有了参照物。2.3 确定清洗目标和边界数据处理脚本不能只考虑当前这一个样本。真正投入项目前需要明确几个边界条件第一清洗对象是字符级别完全重复还是允许中间有空格、标点、大小写差异。如果要清洗你是凑企鹅 你是凑企鹅这类带空格的版本就要在匹配前做空白字符归一化。第二重复单元的最小长度。若阈值设为 1那么“哈哈”会被识别成“哈”的两个重复造成误删。若阈值设得太大又无法覆盖短文本场景。通常中文环境下重复片段最小长度可以从 2 开始例如“你好你好”。第三压缩结果保留哪个部分。连续重复时保留第一次出现的片段还是把两次出现的片段合并成一次这会直接影响下游对文本语义的判读。学习环境里我们可以把这些条件写成脚本参数方便反复调试。生产环境里则要先把参数与业务规则固化再进入自动化流程。3. 从完全重复到相似重复逐层去重的实现思路3.1 第一道防线去掉整行完全重复最基础的操作是整行去重。如果数据来自日志中同一时间点的重复上报或者来自数据库中的冗余读取整行完全重复出现的概率很高。使用 Python 的set可以快速去重但不保留原有顺序。若希望保留第一次出现的顺序推荐使用dict.fromkeysfrom pathlib import Path def dedup_lines_in_file(input_path: str, output_path: str) - None: lines Path(input_path).read_text(encodingutf-8).splitlines() cleaned_lines list(dict.fromkeys(lines)) Path(output_path).write_text( \n.join(cleaned_lines) \n, encodingutf-8, ) print(f清理前: {len(lines)} 行) print(f清理后: {len(cleaned_lines)} 行)这里使用dict.fromkeys而不是set原因是从 Python 3.7 开始 dict 会保留插入顺序能稳定保留第一次出现的行。对于千万级数据量内存占用会偏高更适合用下面的 Linux 命令或者分块文件处理。3.2 第二道防线识别行内连续重复并压缩整行去重处理不了你是凑企鹅你是凑企鹅你是凑企鹅因为这一行只有一个样本。它的问题不是多行相同而是单行内重复了同一个子串三次。下面用“最小重复单元”的思路来检测若整串文本等于某个子串重复 N 次就认为它是一段连续重复文本可以压缩成一次。代码故意从短到长枚举子串长度并且只允许子串完整覆盖全文本。def find_repeat_unit(text: str) - tuple[str, int] | None: 如果 text 由某个子串重复而成返回 (子串, 重复次数)否则返回 None。 n len(text) if n 2: return None for unit_len in range(1, n // 2 1): if n % unit_len ! 0: continue unit text[:unit_len] repeat_count n // unit_len if unit * repeat_count text: return unit, repeat_count return None def compress_line(text: str) - str: unit_info find_repeat_unit(text) if unit_info: unit, _ unit_info return unit return text用这个函数处理标题样本sample 你是凑企鹅你是凑企鹅你是凑企鹅 print(find_repeat_unit(sample)) # (你是凑企鹅, 3) print(compress_line(sample)) # 你是凑企鹅这个做法的核心是从长度为 1 的候选单元开始尝试直到长度超过文本一半。只要某个长度能被整除且按该长度重复后与原文完全相等就找到了完整重复单元。它比正则表达式更可控因为正则很难表达“整串必须由同一个短片段重复 N 次组成”这种约束。实际日志中一行文本可能并不是“整行重复”而是中间夹杂正常内容。此时可以先拆分句子或者把问题弱化为“连续重复区域的识别”。比如某些日志格式为用户提交失败原因超时超时超时请重试若要把中间的连续重复“超时超时超时”压缩成“超时”就要定位重复片段的位置。一个较稳妥的做法是使用正则import re def compress_consecutive_repeats(text: str, min_unit: int 2) - str: 把文本中连续重复 2 次以上的片段压缩为 1 次。 pattern re.compile(r(.{%d,}?)\1 % min_unit) while True: new_text pattern.sub(r\1, text, count1) if new_text text: break text new_text return text需要注意这个正则能处理嵌入在句子里的连续重复但也可能误伤“我跑跑步跑跑步”这类情况。实际使用时要先看清洗目标再决定是否对全文本做这种替换。3.3 用字符熵初步判断文本是否冗余字符熵可以衡量一段文本的信息量。如果文本只有少数几个字符反复出现那么熵值会很低。以随机中文字符组成的句子为例字符种类多且分布均匀熵值会较高。计算文本字符熵的脚本from collections import Counter import math def char_entropy(text: str) - float: if not text: return 0.0 counter Counter(text) total len(text) entropy 0.0 for count in counter.values(): p count / total entropy - p * math.log2(p) return entropy text_repeat 你是凑企鹅你是凑企鹅你是凑企鹅 text_normal 今天天气不错适合户外跑步 print(char_entropy(text_repeat)) print(char_entropy(text_normal))这里有个有趣的坑你是凑企鹅只是把同一个短语重复三次字符频次并没有变化所以整行熵值与单次短语一样。单独看文本字符熵实际上并不总能识别这种“完整短语重复”。因此建议把熵作为辅助信号而不是主判断依据。如果你的样本是AAAAAAAA这种字符级重复熵会明显偏低若是短语级重复还是要结合重复单元算法。3.4 相似文本去重什么时候需要 SimHash 和向量嵌入如果清洗对象不只是完全重复或行内连续重复而是两行文本表达的意思相同但措辞略有不同就应该使用相似度算法。常见做法如下表方法适用规模样本要求优点缺点精确集合去重十万级至千万级整行完全一致简单、快、可解释无法处理近似文本正则压缩任意文本量行内连续重复能压缩行内噪声规则维护成本高SimHash百万级以上文本特征词可提取分桶后可支持大规模去重中文短文本效果不稳定MinHash文档去重集合式文本高召回、可并行需要 Jaccard 相似度阈值编辑距离千条以内精确比较结果准确计算复杂度高向量嵌入任意规模需要模型资源能捕捉语义相似需要 GPU 或向量库学习阶段不需要一上来就上 SimHash。先观察数据如果 90% 的重复都来自“整行完全重复”和“连续拼接”用前两步方法即可清理大部分问题。只有当你发现重复文本中大量夹杂同义词、口语变体和语气词时才考虑引入语义去重。4. 在 Linux 输出端顺手清洗sort、uniq、awk、perl 的组合用法4.1 统计重复次数并删除完全重复行当数据量达到几百万行时Python 的read_text()会把整个文件加载到内存可能触发内存不足。Linux 命令更适合做流式排序和去重。先看每一行的重复次数cd text_cleaner/data sort raw.txt | uniq -c | head -n 20输出示例中计数在最左边代表该行重复次数。sort会改变行的原始顺序所以如果只需要“统计频次”这样最快。如果需要删除重复行并保留首次出现的顺序使用awk数组awk !seen[$0] {print} raw.txt ../output/dedup_lines.txt这个命令的核心是seen[$0]。当遇到第一行时seen[$0]为 0取反后为真于是打印第二次遇到同一行时seen[$0]已经为 1!1为假不再打印。对大量文本而言awk 的处理速度比 Python 脚本快因为它是 C 语言实现的文本处理工具。4.2 处理行内连续重复文本Linux 命令同样可以压缩行内重复。使用 Perl 兼容正则把连续重复的短片段替换成一次perl -CSD -Mutf8 -pe s/(. {2,}?)\1/\1/g ../data/raw.txt ../output/repeat_compressed.txt需要说明的是这类正则对 Unicode 中文支持依赖-CSD和-Mutf8参数而不同平台的 Perl 编译选项可能不同。直接在生产环境执行前先在小样本上测试。如果你更习惯 Python可以把范围控制在脚本里解析速度也足够。命令行方式的优势是无需编写基础设施适合日志出现异常时的快速止血。4.3 用管道一次完成多条规则实际处理时多个规则可以串成管道。下面的例子按顺序完成行内连续重复压缩、去除完全重复行、计算剩余行数perl -CSD -Mutf8 -pe s/(.{2,}?)\1/\1/g raw.txt | awk !seen[$0] | wc -l使用管道的风险在于前面步骤的误伤无法在管道内看到。因此本地临时验证可以这样用但在正式数据清洗任务里每一步的输出都应当保存成中间文件方便检查。5. 怎样验证清洗结果而不是只数删了多少行5.1 对比清洗前后的逐行结果运行清洗脚本后不能只看剩下多少行还要打开文件逐段查看。你可以使用diff命令观察被修改的具体文本diff data/raw.txt output/cleaned.txt对于你是凑企鹅你是凑企鹅你是凑企鹅这行预期 diff 显示的旧版是 18 个字符新版是 6 个字符。这个变化意味着压缩成功。对于本来就正常的文本diff 不应出现任何差异。如果清洗批量很大人工逐行查看不现实需要抽样。写脚本随机抽取清洗前和清洗后的行对让业务同学抽查 100 到 200 条确认没有明显误删。5.2 统计压缩比和误删率在文本清洗任务里有两个指标很有价值。指标一文本压缩率。按行计算清洗后的总字符数除以清洗前的总字符数看整体文本缩减了多少。若压缩率明显低于预期可能是重复内容过多若接近 100%说明当前数据处理步骤几乎没起作用。指标二误删率。人工标记一份 100 条的小样本明确哪些行不应该被修改、哪些短语不应该被压缩。用清洗脚本处理后统计被错误修改的样本比例。from pathlib import Path def compression_ratio(original: str, cleaned: str) - float: len_before len(Path(original).read_text(encodingutf-8)) len_after len(Path(cleaned).read_text(encodingutf-8)) return len_after / max(1, len_before) * 100 print(compression_ratio(data/raw.txt, output/cleaned.txt))这份 100 条小标注集同时也是回归测试集。以后每次修改清洗规则都要重新跑一遍看误删率是否升高。5.3 检查下游分词结果是否恢复正常如果文本最终要进入分词或词频统计可以清洗前后分别跑一遍相同的 jieba 分词代码比较关键词列表。清洗之前你是凑企鹅会以完整片段或异常片段高频出现清洗之后它应当只出现一次不再干扰其他正常词。import jieba from collections import Counter def top_keywords(text: str, top_n: int 10): words [w.strip() for w in jieba.cut(text) if w.strip()] return Counter(words).most_common(top_n) original Path(data/raw.txt).read_text(encodingutf-8) cleaned Path(output/cleaned.txt).read_text(encodingutf-8) print(top_keywords(original)) print(top_keywords(cleaned))如果清洗后关键词分布更接近业务直觉说明去重方向正确。如果某些正常短语被误删则在词频中会看到明显缺失。这也是文本清洗效果最直观的验证方式。6. 常见坑为什么清不干净为什么误删正常文本6.1 只做整行去重漏掉行内重复很多初学者拿到你是凑企鹅你是凑企鹅你是凑企鹅第一反应是交给set或uniq。做完发现数据量没减少因为文件中只有这一行它并不会被整行去重捕获。定位方式打印每条文本的长度观察是否存在长度异常偏长的行。再看是否可以通过“最小重复单元检测”被压缩。处理建议在整行去重之前先做一层行内连续重复压缩。两者是不同粒度的操作不能互相替代。6.2 正则重复阈值太小导致误伤正常表达把正则写成(.{1,}?)\1会把“好好休息”中重复的“好”字识别为连续重复压缩成“好休息”破坏语义。推荐将最小重复单元长度设为 2 或者 3。中文常见叠词“慢慢走”“哈哈笑”很多是单字或双字重复业务中是否需要保留要提前确定。宁可漏过少量脏数据也比把正常语句改错要安全因为漏过的数据还能通过上游策略再处理。6.3 大小写、全半角和不可见字符导致去重失败你是凑企鹅 你是凑企鹅与你是凑企鹅你是凑企鹅从肉眼看很像但一个中间有空格一个没有字符串并不相同。全角逗号和半角逗号、换行符、制表符、零宽空格都可能成为匹配失败的元凶。检查办法是打印文本的repr()或十六进制表示line 你是凑企鹅\u200b你是凑企鹅 print(repr(line)) # 你是凑企鹅\u200b你是凑企鹅处理办法先做 Unicode 归一化将全角字符转为半角、去掉零宽空格再清洗。例如使用标准库unicodedata.normalize(NFKC, text)。import unicodedata def normalize_text(text: str) - str: text unicodedata.normalize(NFKC, text) text text.replace(\u200b, ) # 去掉零宽空格 return re.sub(r\s, , text.strip())6.4 把口语化正常重复也压缩掉了像“是是是你说得对”“好好好我马上看”这类文本连续重复并不是拼写错误而是表达情绪。处理方式有几种。第一种是看重复片段是否构成完整语义单元。“是是是”由单字重复构成通常不算脏数据如果强制压缩成“是”会改变语气。第二种是结合业务语境比如在客服工单中用户重复敲字可能表示强调不一定需要清理。第三种是使用白名单在脚本中保留“好好好”“是是是”“哈哈哈”等常见口语模式。白名单需要持续积累不能一劳永逸。6.5 内存不足或无脑加载整个文件处理超大文本时避免一次性读取整个文件到内存。推荐用生成器逐行处理每读一行、清洗一行、写出到临时文件最后再合并。from pathlib import Path def stream_clean(input_path: str, output_path: str) - None: input_file Path(input_path) output_file Path(output_path) with input_file.open(r, encodingutf-8) as fin, \ output_file.open(w, encodingutf-8) as fout: for line in fin: line line.rstrip(\n) compressed compress_line(line) fout.write(compressed \n)这种方式只需要保存当前行和输出句柄不会一次性占用大量内存。对于 GB 级文件还能结合fileinput或awk分担压力。6.6 清洗结果没有回归验证规则越写越乱清洗规则会随着需求增长而增多。如果每新增一条规则都不重新跑历史样本很容易出现“解决了新问题破坏了旧样本”的情况。常见现象是某天业务方反馈大量正常文本被改成单字而你很难定位是哪条规则导致的。建议维护一个最小回归集包含所有同类问题的代表样本例如整行重复、带空格的连续重复、口语化叠词、URL 参数拼接、日志正常文本。每次修改脚本后运行python scripts/clean.py --input tests/sample_raw.txt --output tests/sample_cleaned.txt diff tests/expected.txt tests/sample_cleaned.txtexpected 文件由人工审核确认只要 diff 为空就可以继续往更大范围的数据上运行。7. 从学习脚本到生产流程可复用清单与扩展方向7.1 清洗规则执行优先级清洗规则不能随机堆叠建议按以下顺序执行能降低误删风险用unicodedata.normalize(NFKC, text)做 Unicode 归一化并去掉零宽空格。将换行、制表符导致的多行文本按字段处理后拆回单行。去掉 URL、HTML 标签、邮箱等明显格式噪声时使用白名单式正则。做整行完全去重保留第一次出现的顺序。做行内连续重复压缩最小重复单元长度建议为 2。根据业务需要加入语气白名单避免误删正常叠词。对长度异常长、重复单元明显的文本单独抽样审核。最后再进入语义级相似度去重并保留原始文本用于审计。这八步并不是严格顺序而是风险从低到高、粒度从不精确到精确的递进。风险越高的步骤越靠后越需要人工抽样检查。7.2 学习环境与生产环境的差别在小数据集上写脚本可以通过read_text()直接完成因为数据不大方便调试。进入生产环境后需要考虑几个因素维度学习/实验环境生产环境数据量几百行到几十万行每天上亿条日志或评论处理方式单机 Python 脚本Spark、Flink、Kafka Streams 等分布式任务规则维护写在 demo 脚本里配置中心或在表结构里保存清洗版本可观测性打印删了多少行输出 metrics、报警、审计日志回滚重新跑一次脚本保留原文件、清洗后文件、规则版本样本验证人工抽查定期运行回归集设置误删率阈值生产环境里文本清洗不只要“删掉重复”还要记录每条文本是在哪个环节、哪条规则下被修改的。比较稳妥的做法是增加一个清洗事件表字段包括原始文本哈希、清洗后文本哈希、规则 ID、处理时间。这样出了问题能回查不至于在黑盒流程中改坏数据。7.3 开发验收前后可以对照的检查清单每写完一版清洗流程可以按这个清单逐项打勾是否保存了原始文本备份防止误删后不可恢复。是否清理了全角、半角、零宽空格的差异。整行完全重复是否已按首次出现顺序去重。是否验证过“行内连续重复”场景而不是只做了整行去重。最小重复单元长度是否设置合理。是否加入口语叠词白名单比如“好好好”“是是是”。是否使用抽样 diff 或人工检查确认没有误删正常文本。是否记录了规则版本和处理数据版本。是否有回归测试集能在规则变更后自动发现破坏。是否确认内存、CPU、输出文件落盘方式满足生产要求。这份清单同样适用于你接手的任何一个历史清洗脚本。先按它逐项检查再决定能不能直接上线能避免很多隐性风险。7.4 从单条样本识别到语义质量监控回到开头的你是凑企鹅你是凑企鹅你是凑企鹅它其实是一个很好的数据质量信号。如果在业务数据里突然大量出现这类“同一短语连续拼接”文本说明上游采集或生成逻辑可能出问题了而不只是需要在清洗阶段花力气修正。更完整的做法是把这段文本的重复度做成一个监控指标。例如抽取每天落库文本的 1%计算“最小重复单元长度”和“压缩比”当异常比例超过设定阈值时触发告警。这会促使团队去排查上游生成逻辑而不是永远在下游打补丁。真正有价值的文本清洗不应该只满足于把脏数据修掉。它应该帮你发现脏数据从哪来并推动上游修复。掌握了重复文本识别、压缩和验证方法后你可以进一步研究文本指纹、SimHash、MinHash 或向量化检索把“字符串去重”升级为“语义级去重”。但无论使用哪种高级方案规则清洗和回归验证依然是不可省略的基础层。