flowchart LR
A[原始 HTML] --> B[提取正文]
B --> C[语言识别]
C --> D[质量过滤]
D --> E[安全过滤]
E --> F[去重]
F --> G[Tokenize & 存储]
第13章 AI 数据流水线
第13章 AI 数据流水线
“Garbage in, garbage out” 是机器学习领域最古老的格言，但在大规模 AI 时代，它不再只是一个警告——它是一个工程学科。
数据流水线是 AI 基础设施的动脉系统。一条设计良好的流水线能让模型团队专注于算法和实验，而一条糟糕的流水线则会让你把 80% 的时间花在”找数据、修数据、等数据”上。本章覆盖从原始数据到训练就绪张量的完整链路。
13.1 数据采集、清洗与去重
为什么采集不是”下载就完事了”
在大规模预训练场景中，数据采集意味着从互联网上抓取数百 TB 的网页文本、代码仓库、学术论文和对话数据。这和业务系统里的 ETL 有本质区别：来源数据是非结构化的、充满噪音的、大量重复的，而且质量分布是长尾的。
Common Crawl、GitHub Archive、arXiv 等公开数据集是预训练数据的主要来源。但直接拿来用会让你吞下大量 SEO 垃圾、机器翻译文本和重复内容。研究表明，数据质量对模型性能的影响远大于数据数量——至少在超过一定规模后如此。
清洗的核心步骤
一个典型的文本清洗流水线包含以下阶段：
提取正文：用 trafilatura 或 resiliparse 从 HTML 中提取正文内容，去除导航栏、广告和模板文本。
语言识别：用 fastText 或 cld3 进行语言分类，低置信度的文本直接丢弃。
质量过滤：这是最关键的一步。常用启发式规则包括：
def quality_filter(text: str) -> bool:
"""基于 CC-Net 规则的简化质量过滤器"""
words = text.split()
if len(words) < 50: # 太短
return False
if len(words) > 100000: # 异常长
return False
# 重复行检测
lines = text.split('\n')
if len(lines) > 1:
dup_ratio = sum(1 for l in lines if lines.count(l) > 1) / len(lines)
if dup_ratio > 0.3:
return False
# 字符质量
alpha_ratio = sum(c.isalpha() for c in text) / len(text)
if alpha_ratio < 0.5:
return False
# 平均词长（英文 3-10 为正常范围）
avg_word_len = sum(len(w) for w in words) / len(words)
if avg_word_len < 2 or avg_word_len > 20:
return False
return True安全过滤：移除包含 PII（个人身份信息）、恶意内容或不适宜文本的文档。
去重：大规模数据的核心挑战
去重是预训练数据处理中计算成本最高的环节。一个 1TB 的语料库，如果做全量两两比较，需要 O(n²) 的计算量——这是不可行的。
实践中采用 MinHash + LSH（Locality-Sensitive Hashing） 方案：
from datasketch import MinHash, MinHashLSH
def build_dedup_index(documents: list[str],
num_perm: int = 128,
threshold: float = 0.8):
"""
构建 MinHash LSH 索引来检测近似重复文档。
num_perm: 排列数，越大越精确但越慢
threshold: Jaccard 相似度阈值，高于此值视为重复
"""
lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)
for doc_id, text in enumerate(documents):
m = MinHash(num_perm=num_perm)
# 用 n-gram 作为 shingle
words = text.lower().split()
for shingle in (tuple(words[i:i+5]) for i in range(len(words) - 4)):
m.update(" ".join(shingle).encode('utf-8'))
# 查询是否已有近似重复
result = lsh.query(m)
if not result: # 没有重复，加入索引
lsh.insert(doc_id, m)
return lsh
Tip
实践建议：对于 TB 级数据，使用 datasketch 的并行化版本或 Spark 上的 MinHash 实现。Google 的 deduplicate-text-datasets 工具链是工业级参考实现。
Warning
去重不是一次性的。你应该在文档级别（document-level）和句子级别（sentence-level）各做一轮。文档级去重移除抄袭和镜像页面，句子级去重防止模型死记特定短语。
13.2 数据格式与高效存储
选对数据格式，训练吞吐量可以差 3-10 倍。这不是微优化——当你的训练成本是每小时数百美元时，格式选择的 ROI 是实打实的。
格式选择决策矩阵
flowchart TD
A[数据特征] --> B{样本大小?}
B -->|< 1MB| C{需要随机访问?}
B -->|> 1MB 视频/音频| D[WebDataset / Sharded]
C -->|是| E[TFRecord / RecordIO]
C -->|否| F[Parquet / Arrow]
D --> G[tar shard + index]
TFRecord
Google 的 TFRecord 本质上是一个长度前缀的 protobuf 序列化流。它的优势在于和 TensorFlow/Keras 生态的深度集成，以及对随机访问的友好（通过 index 文件）。
import tensorflow as tf
def write_tfrecord(samples: list[dict], output_path: str):
"""将样本写入 TFRecord 文件"""
with tf.io.TFRecordWriter(output_path) as writer:
for sample in samples:
feature = {
'text': tf.train.Feature(
bytes_list=tf.train.BytesList(
value=[sample['text'].encode('utf-8')]
)
),
'label': tf.train.Feature(
int64_list=tf.train.Int64List(
value=[sample['label']]
)
),
}
example = tf.train.Example(
features=tf.train.Features(feature=feature)
)
writer.write(example.SerializeToString())
def parse_example(serialized):
"""解析 TFRecord 样本"""
feature_spec = {
'text': tf.io.FixedLenFeature([], tf.string),
'label': tf.io.FixedLenFeature([], tf.int64),
}
return tf.io.parse_single_example(serialized, feature_spec)WebDataset：大文件时代的正确选择
当你的数据是图像、视频或大文本时，把每个样本存成独立文件会导致海量小文件——这在分布式文件系统上是一场灾难。WebDataset 把样本打包成 tar shard（通常 100MB-1GB 每个），并支持顺序读取和并行 shard 处理。
import webdataset as wds
# 写入 WebDataset
def write_webdataset(samples, pattern="shard-%06d.tar", maxcount=10000):
with wds.ShardWriter(pattern, maxcount=maxcount) as sink:
for i, sample in enumerate(samples):
sink.write({
"__key__": f"sample{i:08d}",
"txt": sample['text'].encode('utf-8'),
"cls": str(sample['label']).encode('utf-8'),
})
# 读取 WebDataset（支持并行、shuffle、batch）
dataset = (
wds.WebDataset("shards/shard-{000000..000099}.tar")
.shuffle(1000)
.decode("utf-8")
.batched(32)
)Parquet 与 Arrow：列式存储的胜利
对于表格型数据（特征工程、结构化训练数据），Parquet + Arrow 是黄金组合。Parquet 提供高压缩比的列式存储，Arrow 提供零拷贝的内存表示。
import pyarrow.parquet as pq
import pyarrow as pa
# 写入 Parquet
table = pa.table({
'text': texts,
'label': labels,
'embedding': embeddings, # 嵌套列表也支持
})
pq.write_table(table, 'dataset.parquet', compression='zstd')
# 读取（支持谓词下推、列裁剪）
table = pq.read_table('dataset.parquet', columns=['text', 'label'])
df = table.to_pandas() # 零拷贝转换
Tip
Arrow 的杀手锏：Arrow 的 IPC 格式允许不同进程、不同语言之间零拷贝共享数据。这意味着你的 Python 数据预处理进程和 C++ 训练引擎之间不需要序列化开销。这正是 Ray Data 和 Dask 的底层机制。
格式性能对比
| 格式 | 随机访问 | 顺序吞吐 | 压缩比 | 最佳场景 |
|---|---|---|---|---|
| TFRecord | ✅（需索引） | 高 | 中 | TensorFlow 生态 |
| WebDataset | ❌ | 极高 | 中 | 大文件、多模态 |
| Parquet | ✅ | 中 | 高 | 表格数据 |
| Arrow IPC | ✅ | 高 | 低 | 进程间共享 |
| JSON Lines | ✅ | 低 | 低 | 原型开发 |
13.3 数据预处理与增强
预处理的位置：离线 vs 在线
一个关键的设计决策是：预处理应该在训练前做（离线/预计算）还是在训练中做（在线/实时）？
flowchart TD
A[原始数据] --> B{预处理复杂度}
B -->|低（tokenize, normalize）| C[在线预处理]
B -->|高（增强, 编码）| D[离线预处理]
C --> E[GPU 空闲时计算]
D --> F[预计算后存储]
E --> G[训练循环]
F --> G
原则：如果预处理在 CPU 上耗时不到单步训练时间的 1/10，做在线预处理。否则做离线。
文本数据增强
现代 LLM 训练中最常用的文本增强策略：
import random
import numpy as np
class TextAugmenter:
"""轻量级文本增强，适用于预训练数据"""
@staticmethod
def random_deletion(text: str, p: float = 0.1) -> str:
"""以概率 p 删除每个词"""
words = text.split()
remaining = [w for w in words if random.random() > p]
return ' '.join(remaining) if remaining else words[0]
@staticmethod
def random_swap(text: str, n: int = 2) -> str:
"""随机交换 n 对相邻词"""
words = text.split()
for _ in range(n):
if len(words) > 1:
i, j = random.sample(range(len(words)), 2)
words[i], words[j] = words[j], words[i]
return ' '.join(words)
@staticmethod
def cutmix(text_a: str, text_b: str, alpha: float = 0.5) -> str:
"""将两段文本拼接（适用于预训练）"""
words_a = text_a.split()
cut = int(len(words_a) * alpha)
return ' '.join(words_a[:cut]) + ' ' + text_b图像增强
视觉模型的增强已经高度标准化。albumentations 库提供了工业级的增强流水线：
import albumentations as A
import cv2
# 训练增强流水线
train_transform = A.Compose([
A.RandomResizedCrop(224, 224, scale=(0.08, 1.0)),
A.HorizontalFlip(p=0.5),
A.ColorJitter(brightness=0.2, contrast=0.2,
saturation=0.2, hue=0.1, p=0.8),
A.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]),
A.ToTensorV2(),
])
# 应用
image = cv2.imread("sample.jpg")
image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
augmented = train_transform(image=image)
Warning
增强不是越多越好。过度的数据增强会引入分布偏移——你的训练数据和推理数据来自不同的分布。务必在验证集上不加增强来评估模型。
13.4 流式与批式数据处理
两种范式的区别
批处理（Batch Processing）处理的是静态的、有界的数据集——“把昨天所有的日志跑一遍”。流处理（Stream Processing）处理的是动态的、无界的数据流——“每来一条日志就处理一条”。
AI 数据流水线中两者都有位置：
| 场景 | 推荐范式 | 工具 |
|---|---|---|
| 预训练数据构建 | 批处理 | Spark, Ray Data |
| 在线学习数据喂入 | 流处理 | Kafka + Flink |
| 数据监控与告警 | 流处理 | Flink, Kafka Streams |
| 特征工程 | 批处理 + 流处理 | Feast, Tecton |
Spark 批处理示例
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, udf, countDistinct
from pyspark.sql.types import BooleanType
spark = SparkSession.builder \
.appName("data-pipeline") \
.config("spark.sql.adaptive.enabled", "true") \
.config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
.getOrCreate()
# 读取原始数据
df = spark.read.parquet("s3://raw-data/crawl-2026-07/")
# 质量过滤（UDF）
@udf(BooleanType())
def is_quality(text):
if text is None or len(text) < 200:
return False
alpha_ratio = sum(c.isalpha() for c in text) / len(text)
return alpha_ratio > 0.6
filtered = df.filter(is_quality(col("text")))
# 去重统计
stats = filtered.agg(
count("id").alias("total"),
countDistinct("url_hash").alias("unique_urls")
).collect()[0]
print(f"质量过滤后: {stats.total:,} 文档, 去重率: {1 - stats.unique_urls/stats.total:.2%}")
# 写回（分区存储）
filtered.write \
.mode("overwrite") \
.partitionBy("language") \
.parquet("s3://processed-data/clean-2026-07/")Flink 流处理示例
当你需要实时处理数据（例如用户对话数据实时流入训练系统）：
// Flink Java 示例：实时数据清洗
DataStream<String> stream = env
.addSource(new FlinkKafkaConsumer<>(
"raw-conversations",
new SimpleStringSchema(),
kafkaProps
))
.map(new RichMapFunction<String, CleanedDoc>() {
@Override
public CleanedDoc map(String raw) throws Exception {
// 解析、清洗、过滤
Doc doc = parseJson(raw);
if (isSpam(doc) || containsPII(doc)) {
return null; // 过滤
}
return clean(doc);
}
})
.filter(Objects::nonNull);
// 写入下游存储
stream.addSink(new FileSink<>(...));
Tip
架构建议：大多数 AI 团队应该从批处理开始。只有当你有明确的”数据到模型”延迟要求（如小时级在线学习）时才引入流处理。过早引入流处理会让架构复杂度爆炸。
13.5 数据加载器设计与性能优化
数据加载器是训练的隐形瓶颈
一个常见的性能问题：GPU 利用率只有 50-60%，原因不是模型代码慢，而是数据加载跟不上。GPU 在等数据。
理想的数据加载器应该满足：
flowchart LR
A[存储] -->| prefetch | B[内存缓冲区]
B -->| prefetch | C[GPU/CPU 预处理]
C -->| prefetch | D[GPU 训练]
style A fill:#f9f
style D fill:#ff9
关键思想：流水线化。每个阶段都提前准备好下一阶段需要的数据，让 GPU 永远不需要等。
PyTorch DataLoader 优化
from torch.utils.data import DataLoader, IterableDataset
import torch
class StreamingDataset(IterableDataset):
"""流式数据集，适用于大数据量"""
def __init__(self, shard_paths, tokenizer, max_length=2048):
self.shard_paths = shard_paths
self.tokenizer = tokenizer
self.max_length = max_length
def __iter__(self):
# 每个 worker 处理不同的 shard
worker_info = torch.utils.data.get_worker_info()
if worker_info is None:
shards = self.shard_paths
else:
shards = self.shard_paths[worker_info.id::worker_info.num_workers]
for shard in shards:
for sample in self._read_shard(shard):
tokens = self.tokenizer(
sample['text'],
truncation=True,
max_length=self.max_length,
return_tensors='pt'
)
yield tokens
def _read_shard(self, path):
# 读取 WebDataset / Parquet / TFRecord
import webdataset as wds
for item in wds.Dataset(path).decode():
yield item
# 优化后的 DataLoader
dataloader = DataLoader(
StreamingDataset(shard_paths, tokenizer),
batch_size=32,
num_workers=8, # 多进程预取
pin_memory=True, # GPU 传输优化
prefetch_factor=4, # 每个 worker 预取 4 个 batch
persistent_workers=True, # 避免每个 epoch 重新创建 worker
)关键优化参数解释
| 参数 | 作用 | 推荐值 |
|---|---|---|
num_workers |
并行数据加载进程数 | 物理 CPU 核数的 2-4 倍 |
pin_memory |
页锁定内存，加速 GPU 传输 | True（必须） |
prefetch_factor |
每个 worker 预取的 batch 数 | 2-4 |
persistent_workers |
跨 epoch 保持 worker 存活 | True |
高级技巧：内存映射与零拷贝
对于不需要增强的数据（如已 tokenize 的预训练数据），使用内存映射（mmap）可以完全跳过数据加载开销：
import numpy as np
import mmap
class MMapTokenDataset:
"""使用 mmap 读取预 tokenize 的数据，零拷贝"""
def __init__(self, data_path: str, block_size: int = 2048):
self.block_size = block_size
self.data = np.memmap(
data_path, dtype=np.uint16, mode='r'
)
# uint16 因为 vocab_size < 65536
def __len__(self):
return len(self.data) // self.block_size
def __getitem__(self, idx):
start = idx * self.block_size
end = start + self.block_size
# mmap 不会真的把数据加载到内存
# 操作系统按需 page-in
return torch.from_numpy(
self.data[start:end].astype(np.int64)
)
Tip
生产环境建议：对于 LLM 预训练，最佳实践是把数据预处理成 token 序列，用 np.memmap 存储。Megatron-LM 和 GPT-NeoX 都使用这个方案。GPU 利用率可以稳定在 95% 以上。
Warning
注意 drop_last：分布式训练中，最后一个不完整的 batch 会导致不同 rank 之间的步数不一致。务必设置 drop_last=True。
小结
数据流水线是 AI 基础设施中最容易被忽视、但投资回报率最高的部分。本章覆盖了五个核心环节：
- 采集与清洗——质量过滤和去重是预训练数据的核心挑战，MinHash+LSH 是工业标准方案。
- 数据格式——根据样本大小和访问模式选择 WebDataset（大文件）、Parquet（表格）、TFRecord（TF 生态）或 Arrow IPC（零拷贝共享）。
- 预处理与增强——离线 vs 在线的权衡决定了你的 GPU 利用率。
- 批式 vs 流式——大多数团队应该从批处理开始，只有明确的低延迟需求才引入流处理。
- 数据加载器优化——
num_workers+pin_memory+prefetch_factor是三件套，mmap 是预训练的终极武器。
记住：一个能让 GPU 利用率从 60% 提升到 95% 的数据流水线优化，等价于免费获得了 50% 以上的算力。
延伸阅读
- Data Preparation for Large Language Models: A Survey (Chen et al., 2025) — 最全面的 LLM 数据准备综述
- Deduplicating Training Data Makes Language Models Better (Lee et al., 2022) — 去重对模型性能的量化影响
- WebDataset 文档 —
https://github.com/webdataset/webdataset - PyTorch DataLoader Performance Guide — 官方性能调优指南
- The Pile: An 800GB Dataset of Diverse Text for Language Modeling (Gao et al., 2020) — 数据构建的经典案例研究