Pretraining Data Engineering：从网页到 Training Batch

本章定位：第二批第一优先级。回答一个工程问题：一篇网页是怎么变成 GPU 上的 training batch 的？ 本章不是「爬虫教程」，而是数据工程管线的完整视图——每一站做什么、为什么做、做错的后果是什么。

前置：第 6 章（BPE/Tokenizer）、第 21 章（数据清洗基础概念）。本章在其上建立完整的 pipeline 视角。

18.1 全景管线：从网页到 Batch

Raw Web
  ↓  ① HTML extraction       把网页变成「正文文本」
  ↓  ② Normalization         Unicode / 空白 / 乱码
  ↓  ③ Language ID           这份文档到底是什么语言
  ↓  ④ Quality filtering     低质量内容剔除
  ↓  ⑤ Exact dedup           完全相同文档
  ↓  ⑥ Near dedup            近似重复（MinHash + LSH）
  ↓  ⑦ PII filtering         个人信息
  ↓  ⑧ Benchmark decontam.   评测集泄漏
  ↓  ⑨ Dataset mixture       各领域配比
  ↓  ⑩ Tokenization          文本 → token id
  ↓  ⑪ Document boundary     EOS / mask / packing
  ↓  ⑫ Packing               定长序列、提高利用率
  ↓  ⑬ Sharding              切分文件 + manifest
  ↓  ⑭ Shuffle               全局 / 分片 / 缓冲
  ↓  ⑮ Streaming             按需读取（不装进 RAM）
Training Batch

18.2 Raw Web 与 HTML Extraction

# ❌ 天真做法：拿到的是「页面文字」，不是「正文」
text = soup.get_text()

# ✅ 目标是 main content extraction：识别并保留正文块
# 常见方案：DOM 结构启发式（正文密度、标签路径、段落长度分布）、
#           文本密度算法（如 trafilatura 类方法）、小模型分类

18.3 Normalization：规范化

常见操作：

18.4 Language Identification：语言识别

两个粒度：

18.5 Quality Filtering：质量过滤

常见信号与手段

18.6 Exact Dedup：完全相同去重

import hashlib
def doc_hash(text: str) -> str:
    norm = " ".join(text.split())        # 轻度归一（防空白差异）
    return hashlib.sha256(norm.encode("utf-8")).hexdigest()

seen = set()
for doc in corpus:
    h = doc_hash(doc)
    if h in seen:
        continue          # 完全重复：丢弃
    seen.add(h)
    keep.append(doc)

18.7 Near Dedup：近似去重（MinHash + LSH 全链路）

完整链路（五步）

Document
  ↓ ① shingles   切成 k-gram 集合（如 5-gram 词序列）
  ↓ ② Jaccard    集合相似度 J(A,B) = |A∩B| / |A∪B|
  ↓ ③ MinHash    用 h 个哈希函数压缩成 h 维签名（估计 Jaccard）
  ↓ ④ LSH        签名分桶，同桶内才是「候选对」
  ↓ ⑤ decision   候选对按阈值判定是否重复 → 去重

# 类型：【Skeleton】需要 pip install datasketch
from datasketch import MinHash, MinHashLSH

def minhash_of(text, num_perm=128, k=5):
    m = MinHash(num_perm=num_perm)
    words = text.split()
    for i in range(len(words) - k + 1):
        shingle = " ".join(words[i:i + k])
        m.update(shingle.encode("utf-8"))
    return m

lsh = MinHashLSH(threshold=0.8, num_perm=128)
for doc_id, text in enumerate(corpus):
    lsh.insert(str(doc_id), minhash_of(text))

dups = lsh.query(minhash_of(new_doc))    # 找出与 new_doc 相似的已有文档

18.8 为什么 Dedup 这么重要

18.9 Benchmark Decontamination：去污染

流程

Benchmark sample（问题 + 答案）
  ↓ normalize（与训练数据同一套规范化）
  ↓ n-gram / substring / MinHash 匹配训练文档
  ↓ 命中 → 移除该文档（或该段落）

18.10 PII / Privacy（概念层）

本节只建立概念边界：PII 过滤的目标是降低个人数据被模型记忆与复现的风险。它是数据管线的一站，不是单独的安全合规课程。

18.11 Dataset Mixture：数据配比

18.12 Tokenization Pipeline

Raw Document → tokenizer → token ids（uint16/uint32 存储）

18.13 文档边界：两个不同维度的问题

问题 A：怎么表示 / 隔离文档边界？

如果把 A 文章结尾直接接上 B 文章开头，模型会学到假的跨文档关联（「上一篇的结语」预测「下一篇的标题」）。边界语义的处理方式：

问题 B：怎么把变长文档装进固定长度的训练序列？

两者的关系：先把所有文档连成「Doc A + EOS + Doc B + EOS + …」的流（问题 A 的边界处理），再把流切成定长块（问题 B 的空间利用）。EOS 与 chunk/packing 同时存在——例如：

Doc A
EOS
Doc B
EOS
Doc C ← 同一块里可以包含多篇文档，EOS 标记分界

18.14 定长切块与 Packing：空间利用策略

本课程 Lab 采用：continuous stream + EOS + fixed chunk

step 1：把文档连成 token 流：docA + EOS + docB + EOS + docC + EOS ...
step 2：按 max_seq 连续切定长块（block）
step 3：最后不足一块的 remainder 直接丢弃（drop_last）

def build_token_stream(token_lists, eos_id):
    """连续 token 流：每篇文档后跟一个真实 EOS"""
    stream = []
    for tokens in token_lists:
        stream.extend(tokens)
        stream.append(eos_id)
    return stream

def chunk_stream(stream, seq_len):
    """按 seq_len 连续切块；remainder 由调用方处理（本 Lab drop_last）"""
    return [stream[i:i + seq_len] for i in range(0, len(stream) - seq_len + 1, seq_len)]

正文 + EOS + EOS + EOS + EOS + EOS     ❌

18.15 Sharding：分片

data/
├── shard-00000.bin    # token id 序列（如 uint16）
├── shard-00001.bin
├── ...
└── manifest.json      # 每个 shard 的：token 数 / sha256 / tokenizer 版本 / 领域标签

18.16 Shuffle：打乱

生产中常用「分片 shuffle + 缓冲 shuffle」组合：既保持流式读取，又让局部顺序足够随机。

18.17 Streaming 与 DataLoader 瓶颈

Disk / Object Storage
   ↓ prefetch（提前把下一批搬进内存）
CPU workers（解码 / 拼接 / 打乱）
   ↓ pinned memory（锁页内存，加速 H2D 拷贝）
GPU（计算）

# 类型：【Skeleton】
loader = DataLoader(
    dataset,
    batch_size=32,
    shuffle=True,          # ① 打乱
    num_workers=8,         # ② 并行 worker 预取（关键！）
    pin_memory=True,       # ③ 锁页内存，加速 CPU→GPU 拷贝
    prefetch_factor=4,     # 每个 worker 预取批数
    persistent_workers=True,
)

18.18 实战 Lab：Mini Pretraining Data Pipeline

# 类型：【Runnable】依赖：pip install tokenizers numpy
# mini_pipeline.py —— 迷你预训练数据管线
import hashlib, json, os
import numpy as np
from tokenizers import Tokenizer, models, trainers, pre_tokenizers

# ============ ① normalize ============
def normalize(text: str) -> str:
    text = text.replace("\r\n", "\n").replace("\r", "\n")   # 统一换行
    text = "".join(ch for ch in text if ch == "\n" or ch >= " ")  # 去控制字符
    lines = [" ".join(line.split()) for line in text.split("\n")]  # 折叠空白
    return "\n".join(l for l in lines if l)

# ============ ② exact dedup ============
def exact_dedup(docs):
    seen, keep = set(), []
    for d in docs:
        h = hashlib.sha256(d.encode("utf-8")).hexdigest()
        if h not in seen:
            seen.add(h); keep.append(d)
    return keep

# ============ ③ tokenizer（小语料上现训一个 BBPE） ============
def train_tiny_tokenizer(docs, vocab_size=512, path="mini_tokenizer.json"):
    tok = Tokenizer(models.BPE(unk_token=None))
    tok.pre_tokenizer = pre_tokenizers.ByteLevel(add_prefix_space=False)
    tok.train_from_iterator(docs, trainers.BpeTrainer(
        vocab_size=vocab_size,
        special_tokens=["<|endoftext|>"],
        initial_alphabet=pre_tokenizers.ByteLevel.alphabet(),
    ))
    tok.save(path)
    return tok

# ============ ④ 连续 token 流 + 定长切块（drop_last） ============
def build_token_stream(token_lists, eos_id):
    """连续流：每篇文档后跟【恰好一个】真实 EOS（不做任何填充）"""
    stream = []
    for tokens in token_lists:
        stream.extend(tokens)
        stream.append(eos_id)
    return stream

def chunk_stream(stream, seq_len):
    """按 seq_len 连续切块；返回 (blocks, dropped) —— 不足一块的 remainder 丢弃"""
    n_blocks = len(stream) // seq_len
    blocks = [stream[i * seq_len:(i + 1) * seq_len] for i in range(n_blocks)]
    dropped = len(stream) - n_blocks * seq_len
    return blocks, dropped

# ============ ⑤ shard + manifest ============
def write_shards(blocks, out_dir, seq_len, shard_size=8, dropped=0):
    os.makedirs(out_dir, exist_ok=True)
    manifest = []
    for s, i in enumerate(range(0, len(blocks), shard_size)):
        chunk = blocks[i:i + shard_size]
        arr = np.array(chunk, dtype=np.uint16)
        fname = f"shard-{s:05d}.bin"
        arr.tofile(os.path.join(out_dir, fname))
        manifest.append({
            "file": fname, "sequences": len(chunk),
            "tokens": int(arr.size), "dtype": "uint16",
        })
    with open(os.path.join(out_dir, "manifest.json"), "w") as f:
        json.dump({"shards": manifest, "seq_len": seq_len,
                   "dropped_remainder_tokens": dropped}, f, indent=2)
    return manifest

# ============ 端到端跑一遍 ============
if __name__ == "__main__":
    raw_docs = [
        "Hello world!\n\nThis is doc A.", "Hello world!\nThis is doc A.",   # 归一化后完全相同
        "doc B about 机器学习。", "doc C about transformers and attention.",
        "doc D " * 40,   # 长文档
    ]
    docs = exact_dedup([normalize(d) for d in raw_docs])
    print(f"normalize + dedup: {len(raw_docs)} → {len(docs)} 篇")

    tok = train_tiny_tokenizer(docs, vocab_size=512)
    eos = tok.token_to_id("<|endoftext|>")
    token_lists = [tok.encode(d).ids for d in docs]

    seq_len = 16
    stream = build_token_stream(token_lists, eos)
    blocks, dropped = chunk_stream(stream, seq_len)
    manifest = write_shards(blocks, "mini_data", seq_len, dropped=dropped)

    total = sum(s["tokens"] for s in manifest)
    print(f"token 流总长 {len(stream)}；切成 {len(blocks)} 个定长块（每块 {seq_len}）")
    print(f"Dropped final remainder: {dropped} tokens（drop_last：不足一块的部分主动丢弃）")
    print(f"落盘 {total} token，{len(manifest)} 个 shard")
    print(json.dumps(manifest, indent=2))

验收输出（本课程实测）：

normalize + dedup: 5 → 4 篇
token 流总长 105；切成 6 个定长块（每块 16）
Dropped final remainder: 9 tokens（drop_last：不足一块的部分主动丢弃）
落盘 96 token，1 个 shard

（可自检的语义不变量：流中 EOS 数 = 文档数 = 4；不存在相邻 EOS（说明没有用 EOS 做填充）；每块长度都是 16；丢弃量 = 流长 % 16 = 9。）