做预训练的人都知道一句话垃圾进垃圾出。去年我在MindSpore上跑一个7B规模的稠密模型训练日志里loss漂亮得让人心慌一拉验证集生成的内容里频繁出现重复的广告链、乱码片段和完全说不通的中英混合段落。最后定位到根因不是模型结构不是学习率而是喂给它的预训练文本数据脏得离谱。从那时起我才真正把“数据清洗”从辅助脚本这件事提升到和模型架构同等重要的位置。这篇不是科普文档是我在MindSpore生态里完整跑通一遍预训练文本数据清洗流程后的复盘包含流水线设计、核心算子实现、踩坑实录和一套可以直接拿去改的代码骨架。适合正在做LLM预训练、准备从零搭数据管道、以及被“清洗后效果反而变差”折磨的人参考。1. 为什么文本清洗是预训练的第一道生死线1.1 数据决定模型上限参数只是逼近这个上限大模型圈子里有一句话被反复引用模型架构决定的是“能学多深”数据质量决定的是“底子有多干净”。我自己的实测感受是在固定算力和模型结构下把清洗做扎实带来的收益比把训练步数翻倍还明显。原因是预训练本质上是让模型记住训练集里的统计规律如果统计规律本身包含大量噪声、重复片段、错误字符模型只会把“乱说话”当成一种正常模式去复制。换句话说一个在垃圾文本上训练出来的模型永远不知道“正常文本”长什么样因为它的概率分布里根本没有干净文本的位置。这里有个很现实的指标验证集困惑度Perplexity。假如训练集里混着大量同一个来源的重复句子模型会倾向于输出这段重复文本因为它出现的概率被训练集人为抬高了几百倍。你去看Loss曲线它确实在下降但下降的方向是“越来越擅长复述脏数据”而不是“越来越理解语言结构”。所以清洗不是可有可无的预处理它是预训练的第一道生死线。1.2 预训练文本里的“脏”到底长什么样很多人以为脏数据就是乱码和HTML标签实际远不止。我把线上遇到的高频问题归成五类也是后面清洗算子重点处理的对象格式污染网页抓取残留的HTML标签、CSS代码、JSON脚本、URL链接、特殊符号堆叠。这类问题最常见几乎每个爬虫语料都躲不掉。重复冗余同一个句子在相邻窗口内反复出现或者同一篇文章以不同截断方式重复混入数据。重复数据会严重压缩训练信息的多样性让模型在局部上下文里出现“复读机”现象。低质表达无意义字符序列、纯表情包文本、中英混合但没有任何语义的机器翻译腔、广告文案、骂战内容。这些文本虽然能通过规则过滤但极难完全清除通常需要辅助分类器。内容安全噪声色情、暴力、政治敏感等有害内容。这部分不只是质量维度更是合规的红线。哪怕只是少量漏网都会在模型推理时被恶意放大。编码异常UTF-8乱码、GBK错位、Unicode控制字符、emoji碎片、倒置问号等。这类问题经常被忽略但会把Tokenizer直接搞乱导致训练时出现大量UNK。1.3 清洗的边界不是越干净越好我踩过最大的坑是“过度清洗”。最初我设定了极其激进的过滤阈值把所有包含URL、数字占比过高、长度小于50字的文本全部删掉结果训练出来的模型表达能力急剧下降尤其在代码生成和数学推理上几乎瘫痪。原因很简单代码本身就包含大量URL、特殊符号和数字过度清洗等于把高质量代码语料当垃圾扔了。清洗的目标不是把数据变成“纯净的规范表达”而是把明显无效的噪声去掉保留多样化的有效信息。这个边界需要在清洗后抽一批样本做肉眼检查同时对比清洗前后的训练曲线来摸索。2. 基于MindSpore的清洗流水线整体设计2.1 为什么用MindSpore data pipeline而不是写pandas脚本入行头两年我都是用Python脚本加Pandas做数据清洗一次处理一个文件跑完一个几TB的语料库要花好几天而且CPU利用率极低。后来切到MindSpore的数据管线思路完全变了不要一次性把数据load进内存而是把清洗拆成一个个算子按流式处理。MindSpore的mindspore.dataset模块天然支持这种模式它的GeneratorDataset、TextFileDataset、map、filter、batch等接口可以组合出一条有向无环的数据处理流水线底层自动做多进程并行和算子调度。好处有几点内存占用可控流式处理一个样本处理完就丢弃几百GB语料在单机上也能扛得住。自动并行不需要自己写multiprocessingMindSpore会根据数据集大小和核心数自动分配线程还能和训练进程共享底层的加速能力。和训练闭环清洗完直接生成MindRecord训练时用MindDataset读取整个链路都是原生的不需要中间再转格式。当然不是让你丢掉Pandas。做小规模探索、抽样分析、写一次性统计脚本Pandas还是最快的。但只要数据量上了TB级别就得用算子流式的思路来做。2.2 清洗架构的五个模块我把完整清洗流程拆成五个模块每个模块只负责一件事模块之间通过标准格式对接这样单点出问题可以单独调试数据接入层Ingestion读取原始语料文件夹统一转换成Dataset对象记录来源路径、文件编号等元信息。规则清洗层Rule-based用字符串处理算子完成去标签、去URL、去异常字符、长度过滤、语言识别等动作。去重层Dedup先做精确去重再做MinHash近似去重支持单机多进程和分布式多机两种模式。质量评估层Quality用预训练的RoBERTa或者小号GPT做困惑度打分再用一个轻量分类器过滤低质量文本。该层产出质量分低分样本可以丢弃或降权。格式封装层Export把清洗后的文本按指定结构写入MindRecord文件同时保留元信息用于后续采样和审计。这五层依赖关系清晰上一层输出目录作为下一层输入目录每一层都有独立日志和中断恢复机制。后面详细讲每层的实现。2.3 从原始语料到MindRecord的流转设计有的团队习惯清洗完直接存成纯文本文件训练时再读。我建议不要这么做尤其是语料超过几十GB以后。纯文本文件没有随机索引每次训练都要重新读取整个文件而且无法记录每个样本的元数据和质量分数。更好的方式是生成MindRecord它在MindSpore里相当于一个专业的二进制数据容器支持按索引随机读取配合MindDataset可以直接在训练时做shuffle、repeat、batch不需要每次重复扫描硬盘。流转设计上我会在最后一步把每个原始样本映射成以下结构{ text_id: 12345678, content: 清洗后的纯文本内容, source: web/arxiv/github/..., quality_score: 0.982, dup_count: 1, lang: zh, filtered_by: rule_len/quality/..., raw_path: corpus/2024-01/www.example.com.txt }text_id是全库唯一IDcontent是送入模型的实际文本其余字段不会进入训练但会被保留在MindRecord的伴生字段中。这样做有三个好处样本可溯源训练时可以按source做加权采样或按需屏蔽某类数据质量分数可以联动后面的样本难度采样策略。3. 核心清洗环节实操与代码实现3.1 统一编码与语料读取先解决乱码很多爬虫语料是从各种渠道汇总来的文件编码混乱是常态。我接手的一个数据集里有UTF-8、GBK、GB18030、甚至还有BIG5的残留。如果直接统一用UTF-8去读大批文件会直接抛异常或者静默产生字符。洗数据第一步不是过滤而是把编码统一。在MindSpore里我习惯这样处理先用TextFileDataset按文件读原始二进制再在Python算子中做编码检测和转换。不建议用chardet做全文件检测太慢。我实测下来一个更快的策略是先尝试utf-8严格解码如果失败就再尝试gb18030再失败就尝试big5最后直接丢弃。这个策略在几十个TB的真实语料上正确率超过98%而且速度是chardet的几十倍。import mindspore.dataset as ds from mindspore.dataset import text def decode_bytes(raw): if isinstance(raw, str): return raw, True for enc in (utf-8, gb18030, big5): try: return raw.decode(enc), True except UnicodeDecodeError: continue return None, False def parse_sample(byte_line): line, ok decode_bytes(byte_line) if not ok: return None line line.strip() if not line: return None return line # 按二进制模式读取文件行File预处理为bytes raw_ds ds.TextFileDataset( files_list[/data/corpus/raw/2024/part-0000.txt], shuffleFalse, num_parallel_workers8, shard_equal_rowsTrue, ignore_usageTrue ) clean_ds raw_ds.map(operations[parse_sample], input_columns[text], output_columns[text], num_parallel_workers16)注意ignore_usageTrue这个参数在旧版本里遇到过因为部分文件带有特殊标签导致读取出错的情况打开后能规避。这里parse_sample返回None时MindSpore的map操作默认会丢弃该行这正好用来做粗过滤。3.2 规则过滤快速干掉90%的垃圾规则过滤是整个流程里最朴素但最管用的环节。我用一组正则表达式和长度统计来过滤明显无意义的内容速度极快能够把脏数据的数量级先降下来给后面的深度清理减轻压力。实际用到的规则定义如下连续特殊字符重复三次以上的---、***、等。超长无空格字符序列比如aaaaaaaaaaaaa或者1234567890连续出现40次以上。链接和广告标记包含http://、www.、[广告]、免费领取等关键词且占比过高。句子数量过少的文本正常文章至少包含5个标点符号低于阈值可直接丢弃。内容长度异常过短少于20个字符和过长超过10000字符但未分句都需要进一步处理。过短大概率是碎片信息过长的文本建议做截断而不是整体丢弃防止溢出。下面这段代码是我在map算子里的核心规则函数import re URL_PATTERN re.compile(r(https?://|www\.)[\w\-./?%#], re.I) BAD_CHAR_PATTERN re.compile(r[^\S\n]{5,}) REPEAT_SYMBOL_PATTERN re.compile(r([\-*_#~]){5,}) def rule_filter(text, min_len20, max_len8000): if text is None: return False if len(text) min_len or len(text) max_len: return False if REPEAT_SYMBOL_PATTERN.search(text): return False url_count len(URL_PATTERN.findall(text)) if url_count max(2, len(text) // 200): return False chinese_count sum(1 for ch in text if \u4e00 ch \u9fff) alpha_count sum(1 for ch in text if ch.isalpha()) if chinese_count 0 and alpha_count 0: return False # 连续无意义字母/数字串检测 if re.search(r[a-zA-Z0-9]{50,}, text): return False return True clean_ds clean_ds.filter(predicaterule_filter, input_columns[text])这里的关键不是正则写得有多完美而是过滤阈值要宽松。我一开始把URL数量设为0结果大量代码和技术博客被误杀。后来改成相对占比误杀率显著降低。规则清洗的操作顺序也很重要先用最简单、计算量最小的过滤比如长度、空行把明显垃圾丢掉再用正则这类中等开销的逻辑处理最后再用模型分类器做深度过滤。这样可以避免把昂贵的模型推理浪费在本来就是垃圾的样本上。3.3 精确去重与近似去重MinHash在MindSpore里的落地预训练语料里的重复问题是隐形的杀手。一个爬虫站点可能把同一篇文章抓了几十遍只是URL不同。如果你不去重模型会认为“知乎”是全网最高频的中文词因为同一个回答被重复灌进去5000次。传统精确去重很简单直接对整篇文本取哈希丢进set里判断。但真实场景里更麻烦的是近似重复文本被插入了几个无关字符、改了标题、加了尾巴哈希就完全不同了。我采用的方案是两级去重第一级是精确去重用文本的SHA-1哈希作为指纹只保留第一次出现的样本。这个操作消耗极小能去掉至少30%的显式重复。第二级是MinHash近似去重将文本切分成若干个词级别的shingle比如连续5个词作为一个窗口然后对每个shingle做哈希取窗口内最小的K个哈希值作为该文本的签名最后比较签名的Jaccard相似度来判断是否近似重复。在MindSpore里实现MinHash不需要引入额外数据库驱动。我们可以把整个去重过程做成一个map算子在算子内部维护一个全局的哈希集合然后输出是否保留。但要注意MindSpore的map算子天然是做数据流并行的如果你在普通Python函数里放一个全局set多进程环境下每个进程都会维护自己的副本无法做到全局去重。我踩过这个坑一开始在map里直接用全局set跑了100个进程每个进程各去重各的最终重复率只下降了40%等于没做。正确写法是用GeneratorDataset配合一个串行的自定义迭代器或者将整个去重任务独立成一个Python脚本分片后用布隆过滤器汇总去重结果。为了简洁我把去重核心逻辑写成一个伪代码级别的方案from datasketch import MinHash def minhash_sig(text, num_perm128): tokens text.split() if len(tokens) 4: return None m MinHash(num_permnum_perm) for i in range(len(tokens) - 4): shingle .join(tokens[i:i4]) m.update(shingle.encode(utf-8)) return m # 分片处理每个分片维护自己的集合 seen_hashes set() seen_minhash [] def dedup_op(text): exact_hash hashlib.sha1(text.encode(utf-8)).hexdigest() if exact_hash in seen_hashes: return None seen_hashes.add(exact_hash) sig minhash_sig(text) if sig is None: return text for old in seen_minhash: if jaccard(sig, old) 0.85: return None seen_minhash.append(sig) return text这段代码只适合单机小规模验证真实大规模去重时我会用datasketch的MinHashLSH做分桶并配合Redis保存全局指纹。不过核心思想不变先用精确去重快速降量再用MinHash找出那些“形不同但神似”的文本。3.4 质量评估困惑度打分与分类器过滤规则过滤和去重能解决结构性问题但无法判断文本“是否自然”。比如一段话“我爱你爱着你就像老鼠爱大米”重复了200遍规则过滤完全拦不住它没有乱码也不是超长但它绝对是垃圾。针对这类问题我引入了质量评估层分成两个组件第一个组件是困惑度打分。用一个在开源语料上预训练好的小模型我用的是RoBERTa-wwm-ext-small对每条文本计算困惑度。困惑度是指模型认为这段文本“意外”的程度真实文本的困惑度会落在某个范围内而重复、乱写、机器翻译腔的文本困惑度会明显异常。实际操作时我并不是对每个样本都做完整推理而是先抽样一部分算分确定一个合理的PPL阈值再全量跑。import numpy as np import mindspore as ms from mindspore import ops from transformers import AutoTokenizer, AutoModelForMaskedLM tokenizer AutoTokenizer.from_pretrained(hfl/rbt3) model AutoModelForMaskedLM.from_pretrained(hfl/rbt3) def ppl_score(text): inputs tokenizer(text, return_tensorspt, truncationTrue, max_length128) with torch.no_grad(): outputs model(**inputs, labelsinputs[input_ids]) loss outputs.loss.item() return 2.71828 ** loss # exp(loss) # 在MindSpore数据流中只对通过规则过滤的样本执行 def quality_filter(text, ppl_threshold150): score ppl_score(text) return score is not None and score ppl_threshold这里我实际上混用了PyTorch的模型来做离线打分转换回MindSpore训练时并不冲突因为这是数据预处理阶段不上模型训练链路。如果你更追求端到端MindSpore可以换成mindspore.nn.transformer里的Encoder来做打分但没必要离线打分工具链用成熟的开源模型会更省事。第二个组件是分类器过滤。困惑度能抓“不自然”但不能抓“虽然自然但是有害”的文本。我会额外维护一个轻量分类器训练语料来自人工标注的样本标签有三类正常、低质、敏感。分类器不需要太复杂一个简单的MLP或者TextCNN就够用输入用Tokenizer编码后的首尾拼接特征。最终质量评分q 0.7 * ppl_normalize 0.3 * classifier_prob_normal低于0.5的样本直接过滤。这个评分还会写进MindRecord后续采样时可以做困难样本挖掘。3.5 自定义算子与MindRecord生成把清洗结果变成训练能吃的格式前面所有清洗结果现在还是一个个Dataset里的文本行。为了训练时能高效读取我会用MindSpore的MindRecord格式把数据固化下来。这个环节也是坑最多的地方一定要用map的output_columns把元数据一起带出来。先定义写入函数import mindspore.dataset as ds from mindspore.mindrecord import FileWriter def write_mindrecord(clean_ds, output_pathdata/train.mindrecord): writer FileWriter(output_path, shard_num4) schema_json { text_id: {type: int64}, content: {type: string}, source: {type: string}, quality_score: {type: float32}, lang: {type: string} } writer.add_schema(schema_json, Pre-training corpus after cleaning) for sample in clean_ds.create_dict_iterator(output_numpyTrue): writer.write_raw_data([sample]) writer.commit()这里有两个细节。第一shard_num决定了生成的MindRecord会被分成几个分片建议跟训练时的卡数成比例比如8卡训练就设8读取时每张卡各读一个分片避免所有卡都去抢同一份文件。第二create_dict_iterator输出的是numpy数组写入时一定要转成对应schema类型否则会报类型不匹配。如果清洗后的Dataset非常大直接create_dict_iterator还是有内存风险。稳妥的做法是边处理边写入不要等整个Dataset物化。你可以在write_mindrecord函数内部对clean_ds做一个for循环逐批次写入写完一个批次就释放一批内存。实际写几TB语料时内存占用量可以压到4GB以内完全可控。4. 常见问题与排查技巧实录4.1 数据流OOM和线程爆炸第一次跑完整流水线时我在单台32核机器上处理1TB语料跑了一个多小时直接OOM。排查后发现是num_parallel_workers设得过高同时TextFileDataset默认会把每个文件的所有内容读入缓冲区。MindSpore的数据管线用起来很有迷惑性你以为是在流式处理但某些API内部会把数据做预取当prefetch_size过大的时候内存是瞬间被填满的。我的统一解决方案是把num_parallel_workers控制在核心数的一半以内prefetch_size从默认值改成2或4并且每种复杂算子比如PPL打分单独限制并发数。另外清洗过程分阶段落盘不要试图在一个Dataset里把所有算子串起来后整体物化那样等于把整条数据管线同时加载进内存。正确的做法是每完成一个阶段就把中间结果写到磁盘下一阶段再读取。中间结果虽然会占用一些磁盘IO但换来了稳定的内存表现。4.2 清洗后数据分布漂移导致效果变差最困惑的一次是清洗完以后训练loss比清洗前还高。检查后发现过滤比例太强把很多长尾内容全部删掉训练集分布和真实世界里人类说话的方式产生了巨大偏移。模型在训练时看到的信息过于“规范”反而失去了泛化能力。后来我每次清洗都会在流水线最后抽样500条样本做人工检查统计被删除样本的类型和比例。如果某个领域比如代码、诗歌、方言的删减比例明显高于平均水平就要主动调低对应规则的强度。这么做之后训练loss虽然不会立刻降到更低但生成质量的主观感受和下游评测指标明显更稳。4.3 正则过滤误杀高价值文本直接做文本过滤时我一开始只用了一个简单的规则如果文本包含http://就直接丢弃。结果导致大量代码注释、博客附录、Markdown文档被误杀。训练出来的模型在涉及编程类问题时频繁“失忆”。最后改成了相对占比判断并且对包含代码块标记的文本单独放宽规则。千万不要在清洗初期用“零容忍”策略那是把双刃剑。你需要先跑通流程再逐步收紧规则每一次规则改动后都要重新评估删减样本的构成。4.4 多卡训练无法读取MindRecord文件我在多机多卡训练时遇到过一个问题MindRecord文件正常生成但训练开始后其中两张卡一直卡住没有任何报错。排查发现是文件分片数量小于训练卡数导致部分卡分不到分片MindDataset一直在等。解决方案有两种要么把shard_num设置为卡数的整数倍要么在读取时用num_shards和shard_id参数手动指定每张卡读取的分片范围。更推荐后者因为生成文件时你不一定知道最终训练卡数。train_ds ds.MindDataset( dataset_files[/data/train.mindrecord0, /data/train.mindrecord1], num_shardsrank_size, shard_idrank_id, shuffleTrue )同时shuffle参数要在MindDataset里设置而不要在后续batch之前再map一个shuffle操作那样既慢又容易让不同卡的数据交错。4.5 清洗流水线中断恢复清洗几百TB数据不是一次能跑完的真跑起来经常遇到断电、磁盘满、程序崩溃。最开始我犯过傻用shell脚本保存每个阶段的临时目录中断只能从头开始。后来优化成用Meta文件记录每个阶段已处理的文件列表重启时跳过已完成的文件。MindSpore的Dataset本身不提供检查点你自己在DataFrame或者dict层面做文件级记录就行。我的实现很简单每处理完一个原始文件就往一个done.txt里追加一行文件名。重启时读这个文件生成exclude集合TextFileDataset传入文件列表时过滤掉已完成文件即可。这样中断恢复的成本几乎为零。4.6 验证清洗效果的小技巧清洗效果不能只盯着训练loss看还要对比清洗前后模型在一些小样本上的生成质量。我的习惯是每清洗一版语料就用同一个随机种子、相同模型结构跑一个缩略版预训练大概几十万步然后跑一组通用NLP评测任务文本分类、抽取式摘要、类似C-Eval那种选择题。如果评测分数确实提升了再投入全量训练。这样做虽然会多花点算力但能避免反复清洗后才发现方向错了带来的巨大浪费。还有一个小技巧在清洗后的语料里专门保留一部分“高难样本”训练完以后单独观察模型在这些样本上的困惑度表现这些样本往往比总验证集更灵敏。写在最后的经验做了大半年的数据清洗复盘我最大的体会是清洗流程的价值不在于某个算法多炫而在于每一层过滤都能说得清楚“为什么这么过滤”以及“删掉这些样本对模型有什么影响”。MindSpore的数据管线给了一个很舒服的载体让我可以用算子化、流式化的方式把清洗逻辑组织成一条能回溯、能断点续跑、能并行的流水线而不是靠一堆临时脚本去处理积木。如果你正在搭自己的预训练数据清洗流程我的建议是先拿10GB左右的样本把整条流水线跑通生成一份清洗前后对比报告再投到全量数据上。不要一上来就写几百个规则先掌握编码统一、规则过滤、去重、质量评估和MindRecord导出这五个核心环节后续加规则、加模型、加分布式都只是锦上添花。最后分享一个小技巧清洗过程中定期把“被过滤掉的样本”而不是“保留的样本”抽样打印出来看看。你的直觉往往会从这些被删掉的样本中获得最多信息比看统计指标更快发现规则误杀或者过滤不足的问题。这个习惯我保留到现在每次清洗版本迭代都靠它把方向校准到正确的位置。