BERT预训练数据集构建指南：从数据清洗到高效分布式处理

1次阅读

没有评论

共计 2716 个字符，预计需要花费 7 分钟才能阅读完成。

构建 BERT 预训练数据集时，首先需要明确数据的核心要求。这些要求直接影响模型性能和训练效率。

文本长度分布 ：BERT 的输入限制为 512 个 token，数据应包含不同长度的文本，从短句到长段落。理想分布是：
20% 文本 <64 tokens
50% 文本 64-256 tokens
30% 文本 256-512 tokens
领域覆盖 ：为了模型的通用性，数据应来自多个领域：
新闻（30%）
百科（25%）
论坛（20%）
学术（15%）
其他（10%）
数据质量 ：
重复率 <5%
非文本内容 <1%
语言一致性 >95%

高质量的数据清洗流水线是预训练成功的基础。以下是关键步骤：

去重：
精确去重（完全相同的文档）
模糊去重（相似度 >90% 的文档）
使用 SimHash 算法进行高效去重
标准化 ：
Unicode 规范化（NFKC）
全角转半角
统一标点符号
去除不可见字符
敏感信息过滤 ：
正则表达式匹配隐私信息（如身份证号、电话号码）
关键词黑名单过滤
使用预训练模型检测不当内容

使用 HuggingFace Tokenizer 时，可以通过以下方式优化：

并行处理 ：

from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
tokenizer.backend_tokenizer.enable_parallelism()

批处理 ：

texts = ["sample text 1", "sample text 2"]
batch_encodings = tokenizer(texts, truncation=True, padding='max_length', max_length=512, return_tensors="pt")

缓存：
预处理结果保存到磁盘
使用 memory mapping 加速后续加载

对于大规模数据，Apache Beam 提供高效的分布式处理能力：

import apache_beam as beam

class DataPreprocessing(beam.DoFn):
    def process(self, element):
        # 数据清洗和分词逻辑
        yield processed_element

with beam.Pipeline() as p:
    (p | 'ReadData' >> beam.io.ReadFromText('input.txt')
       | 'Preprocess' >> beam.ParDo(DataPreprocessing())
       | 'WriteData' >> beam.io.WriteToText('output'))

内存映射格式可以极大提高数据加载效率：

import numpy as np

# 保存为 numpy 内存映射格式
def save_memmap(data, filename):
    arr = np.array(data, dtype=np.int32)
    np.save(filename, arr)
    mmap = np.load(filename, mmap_mode='r')
    return mmap

# 加载示例
mmap_data = save_memmap(batch_encodings['input_ids'], 'bert_data.npy')

处理多语言数据时，平衡是关键：

使用 fastText 进行语言检测
按语言比例采样
动态调整批次中的语言分布

BERT 的掩码策略直接影响训练效果：

15% 的 token 被掩码
其中 80% 替换为 [MASK]，10% 随机替换，10% 保持不变
使用 N -gram 掩码提升难度

大规模训练时，checkpoint 策略很重要：

每 10000 步保存一次
使用分布式文件系统（如 HDFS）
保存优化器状态

数据泄露 ：
严格分离训练 / 验证 / 测试数据
检查重叠文档
内存爆炸 ：
使用生成器而非列表
限制单批次大小
分词器不一致 ：
预训练和微调使用相同分词器
检查特殊 token 是否一致

以下是可复用的数据处理类：

import os
import numpy as np
from transformers import AutoTokenizer

class BERTDataProcessor:
    def __init__(self, model_name='bert-base-uncased'):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.tokenizer.backend_tokenizer.enable_parallelism()

    def preprocess_text(self, text):
        # 文本清洗逻辑
        return cleaned_text

    def tokenize_batch(self, texts, max_length=512):
        return self.tokenizer(
            texts,
            truncation=True,
            padding='max_length',
            max_length=max_length,
            return_tensors="pt"
        )

    def save_to_memmap(self, encodings, output_dir):
        os.makedirs(output_dir, exist_ok=True)
        input_ids_path = os.path.join(output_dir, 'input_ids.npy')
        attention_path = os.path.join(output_dir, 'attention_mask.npy')

        np.save(input_ids_path, encodings['input_ids'].numpy())
        np.save(attention_path, encodings['attention_mask'].numpy())

        return {'input_ids': np.load(input_ids_path, mmap_mode='r'),
            'attention_mask': np.load(attention_path, mmap_mode='r')
        }

关键指标需要持续监控：