共计 2938 个字符，预计需要花费 8 分钟才能阅读完成。
背景痛点：为什么你的 BERT 训练总是效果差？
刚开始接触 BERT 预训练时，我遇到过模型效果不稳定的问题。后来发现，90% 的问题都出在数据集上。原始文本数据常见的坑包括：
- 编码混乱：混合了 UTF-8、GBK、ISO-8859- 1 等多种编码，导致读取时出现乱码
- 噪声数据：包含 HTML 标签、特殊符号、广告文本等无效内容
- 标注缺失：部分语料缺少段落分隔标记，影响 NSP 任务效果
- 数据倾斜：某些领域文本占比过高（如新闻数据中政治类占 70%+）
这些会导致模型在下游任务中出现：MLM 准确率波动大、文本生成不连贯、领域适应能力差等问题。
技术方案：四步构建工业级数据集
1. 数据采集：找对源头事半功倍
推荐这些高质量公开数据集：
- 中文：维基百科 dump、人民日报语料、CLUE 基准数据集
- 英文：BookCorpus、Common Crawl、StackExchange 存档
如果必须爬取数据，注意：
# 遵守 robots.txt 规则的最佳实践
import requests
from bs4 import BeautifulSoup
def safe_crawler(url: str) -> str:
try:
resp = requests.get(url, timeout=10, headers={'User-Agent': 'ResearchBot/1.0'})
resp.raise_for_status()
return BeautifulSoup(resp.text, 'lxml').get_text()
except Exception as e:
print(f"Error crawling {url}: {str(e)}")
return ""2. 数据清洗：从垃圾堆里淘金子
核心清洗流程：
- 统一编码：用 chardet 检测后转 UTF-8
- 去除噪声：正则表达式处理
- 文本规范化：全角转半角、繁体转简体
import re
import zhconv
def clean_text(text: str) -> str:
# 去除 HTML 标签
text = re.sub(r'<[^>]+>', '', text)
# 合并连续空白符
text = re.sub(r'\s+', ' ', text)
# 繁体转简体（中文数据集用）text = zhconv.convert(text, 'zh-cn')
return text.strip()3. 分词处理：WordPiece 实战
BERT 采用 WordPiece 分词，与 BPE 的主要区别：
- WordPiece 基于概率合并，BPE 基于频次
- WordPiece 的 ## 前缀表示子词
使用 HuggingFace Tokenizer 快速实现：
from tokenizers import Tokenizer
from tokenizers.models import WordPiece
from tokenizers.trainers import WordPieceTrainer
tokenizer = Tokenizer(WordPiece(unk_token="[UNK]"))
trainer = WordPieceTrainer(
vocab_size=32000,
special_tokens=["[PAD]", "[UNK]", "[CLS]", "[SEP]", "[MASK]"]
)
# 传入清洗后的文本文件路径
tokenizer.train(files=["cleaned_data.txt"], trainer=trainer)
tokenizer.save("wordpiece.json") # 保存分词模型4. TFRecord 生成：分布式处理大文件
将文本转换为 TFRecord 格式，关键技巧：
- 使用 sharding 避免单个文件过大
- 并行化处理加速转换
import tensorflow as tf
from multiprocessing import Pool
def create_tf_example(text: str, tokenizer) -> tf.train.Example:
tokens = tokenizer.encode(text).ids
features = {"input_ids": tf.train.Feature(int64_list=tf.train.Int64List(value=tokens)),
"length": tf.train.Feature(int64_list=tf.train.Int64List(value=[len(tokens)]))
}
return tf.train.Example(features=tf.train.Features(feature=features))
def process_shard(shard_path: str):
with tf.io.TFRecordWriter(shard_path) as writer:
for text in load_text_chunk(): # 你的数据加载函数
example = create_tf_example(text, tokenizer)
writer.write(example.SerializeToString())
# 并行处理 8 个分片
with Pool(8) as p:
p.map(process_shard, [f"data-{i}.tfrecord" for i in range(8)])避坑指南：血泪经验总结
内存优化：处理 100GB+ 文本的秘诀
- 使用生成器逐行读取文件
- 分 chunk 处理（建议每个 chunk 10 万行）
- 及时释放内存：
def chunk_processor(file_path: str, chunk_size: int = 100000):
with open(file_path, 'r', encoding='utf-8') as f:
chunk = []
for line in f:
chunk.append(clean_text(line))
if len(chunk) >= chunk_size:
yield chunk
chunk = []
if chunk: # 处理剩余行
yield chunk分布式训练：Input Pipeline 配置
在 TF2.x 中优化数据读取：
- 使用
tf.data.Dataset.interleave并行 IO - 设置
prefetch重叠计算与数据加载 - 合理设置
num_parallel_calls（通常设为 CPU 核心数）
dataset = tf.data.TFRecordDataset(
filenames,
num_parallel_reads=tf.data.AUTOTUNE
).shuffle(buffer_size=10000).batch(256).prefetch(10)性能验证：数据规模的影响
测试环境：V100 GPU, Batch Size=256
| 数据量 | 清洗耗时 | 训练速度（steps/sec） |
|---|---|---|
| 1GB | 2min | 15.2 |
| 10GB | 18min | 14.8 |
| 100GB | 2.5h | 14.1 |
可以看到，当数据量增长 100 倍时，训练速度仅下降 7%，说明我们的 pipeline 具有良好的扩展性。
开放性问题
当你费尽心思构建好数据集后，不妨思考：与直接使用公开数据集相比，自建数据集在哪些场景下更有优势？领域适配性和数据多样性之间应该如何权衡？欢迎在评论区分享你的见解。
完整可执行代码已上传 Colab：点击打开笔记本
正文完
