第13章 存储与数据工程

本章覆盖 AI 训练与推理场景的存储系统设计与数据工程全链路。前半部分讨论存储基础设施：I/O 特征、并行文件系统、 对象存储、检查点与向量数据库等专项存储，以及一套训练存储方案的完整设计实践。后半部分讨论数据工程：预处理管 线、分片加载、质量过滤与配比、版本管理、多模态处理与 PB 级管线实战。存储硬件选型细节与万卡集群网络拓扑不在 此展开。

13.1 存储与 I/O 特征

AI 工作负载的 I/O 模式与传统 HPC 和企业应用有本质差异。理解各阶段的数据流动特征是设计 AI 存储系统的基础。从数 据采集到推理服务，整个生命周期的 I/O 特征经历了从“大块顺序”到“小块随机”再到“混合爆发”的剧变。

13.1.1 数据生命周期 I/O 特征

1. 数据摄入阶段 数据摄入（Data Ingestion）阶段从互联网、传感器、数据库等来源采集原始数据，其 I/O 特征如下： •访问模式：大规模顺序写入（数 TB 到数 PB 级数据集） •I/O 大小：典型 1-64 MB 的块级写入 •并发度：中等（数十到数百个并行写入流） •延迟要求：宽松（秒级即可接受） •存储层：对象存储（S3/MinIO）或分布式文件系统作为 landing zone 以 Common Crawl 数据为例，每月新增约 300 TB 的 WARC 原始数据，写入速率约 100-200 MB/s，远低于高性能存储系 统的带宽上限。这一阶段的瓶颈通常在网络入口带宽而非存储本身。
2. 数据预处理阶段 数据预处理（Tokenization、Filtering、Deduplication）是存储压力最大的阶段： •访问模式：随机读取（读取分散的文本片段）+ 顺序写入（输出 tokenized 序列） •I/O 大小：读取以 4 KB-1 MB 为主（小文件/块级随机），写入以 16 KB-4 MB 为主 •并发度：极高（数千到数万个并发任务，对应 Spark/Ray 的并行 partition 数） •延迟要求：中等（毫秒级小 I/O 延迟，大 I/O 秒级可接受） •存储需求：需要同时支持高 IOPS（随机读）和高吞吐（顺序写） 典型场景：清洗 10 TB 文本数据产生 500 GB tokenized 数据。在 1000 个并发 worker 下，每个 worker 需处理约 10 GB 数据，随机 I/O 压力极大。使用 SSD 或 NVMe 作为预处理工作目录，通过网络文件系统（NFS/SMB）或并行文件系统 （Lustre/GPFS）共享中间结果。
3. 模型训练阶段 训练阶段的 I/O 有三个独立的数据流，其特征差异显著，如表13-1所示。 表13-1 训练阶段数据流I/O特征 数据流 访问模式 典型规模 存储介质 延迟档位 数据加载 顺序/伪随机读（shard 级） 批次级 1-16 MB/GPU NVMe 本地缓存 + 并行文件 50-200 ms（预取窗 系统 口） 数据流 访问模式 典型规模 存储介质 延迟档位 检查点写 大规模突发顺序写 单模型约 1.1-2.8 TB（70B-175B SSD 并行文件系统 30-120 s 窗口 入 全量） 日志与指 高频小写 KB 级 本地 NVMe / 日志聚合系统 秒级容忍 标 检查点写入是训练阶段存储压力的主要来源，对聚合写带宽的要求是训练存储规划的核心约束。数据加载的单步净读入仅 数十 MB 量级，但考虑随机访问模式、缓存未命中与并发，文件系统仍按数百 MB/s 至数 GB/s 的聚合带宽规划。物理读 带宽不是瓶颈，元数据与 IOPS 压力才是。
4. 模型推理阶段 推理服务的 I/O 特征相对稳定： •模型加载：启动时一次性大块顺序读取（模型权重，1 GB-数百 GB） •KV Cache：运行时高频小块读/写（每 token 生成需读取/更新 KV Cache 条目） •请求日志：持续低量写入 模型加载的延迟直接影响冷启动时间。在 Serverless 推理中，模型从对象存储拉取到 GPU 的时间是核心冷启动延迟来 源。例如 Llama-70B（约 140 GB）从 S3 通过 25 Gbps 网络下载约需 45 秒，从本地并行文件系统约需 5-10 秒。
5. 存储分层策略 训练数据在整个生命周期按热度分布于不同存储层级，形成热、温、冷三层架构，如图13-1所示。 Checkpoint Storage Cold Storage Checkpoint Parallel File System HDD/S Archive Object Store Glacier SD Raw Data ETL Preprocessed Data Training Load GPU Node Object Store S3/MinIO Parallel File System SSD Local NVMe Cache Inference Deploy Model Repository Model Pull Inference Node Object Store GPU HBM + CPU RAM 图13-1 AI数据流与存储分层架构 各层的设计要点： •热层（GPU HBM + NVMe Cache）：延迟 < 1 μs（HBM）/ < 100 μs（NVMe），带宽可达 TB/s •温层（并行文件系统 SSD）：延迟约 1 ms，聚合带宽 100+ GB/s •冷层（对象存储 HDD）：延迟约 10-100 ms，带宽 GB/s 级，成本最低（$0.02/GB-month，SSD 约 $0.10/GB-month） 不同阶段的 I/O 指标需求差异显著，归纳如表13-2所示。 表13-2 各阶段I/O指标与推荐存储 AI 阶段 主要 I/O 模式 IOPS 需求 带宽需求 延迟容限 推荐存储介质 数据摄入 大块顺序写 低 100 MB/s - 1 GB/s 秒级 HDD 对象存储 预处理 混合读写 高 (10K-100K) 1-10 GB/s 毫秒级 NVMe SSD 训练数据加载 顺序/伪随机读 中 数百 MB/s-数 GB/s 聚合 50-200 ms NVMe + 并行 FS 检查点写入 大块突发写 低 50+ GB/s（规划值） 30-120s 窗口 SSD 并行 FS 推理模型加载 大块顺序读 低 1-10 GB/s 5-60s SSD 或 对象存储 KV Cache I/O 随机小块读/写 高 0.5-2 GB/s per GPU < 10 μs GPU HBM / CPU RAM 理解这些 I/O 特征是后续选择存储架构、设计检查点方案和配置 KV Cache 卸载策略的前提。

13.1.2 并行文件系统对比

并行文件系统（Parallel File System）是 AI 训练集群存储基础设施的核心。与 NFS 等单出口 NAS 系统不同，并行文件系 统通过将数据分散到多个存储服务器（OSS/NSD），实现聚合带宽的近似线性扩展。以下从架构特点、性能基准和适用性 三个维度，对比主流并行文件系统在 AI 场景下的表现。

1. Lustre Lustre 是全球 TOP500 超算中使用最广泛的并行文件系统，也是 OpenAI、Meta、DeepMind 等 AI 主力团队的存储选 择。 架构组件： •MDS（Metadata Server）：管理文件元数据（目录结构、权限、文件属性），一般 1-2 个 active-standby 对 •MDT（Metadata Target）：元数据的物理存储，NVMe SSD 推荐 •OSS（Object Storage Server）：存储数据内容，典型 4-16 台 •OST（Object Storage Target）：数据物理存储，每台 OSS 管理 2-8 个 OST •Client：运行在 GPU 节点上的 Lustre 客户端内核模块 核心机制： •分条（Striping）：文件被分割为固定大小的分条单元，轮流分布到多个 OST 上，分条数（stripe count）和分条大小 （stripe size）是关键调优参数 •客户端缓存：client-side cache 将热点数据缓存在 GPU 节点的 Page Cache 中，减少 OSS 访问 •LNet：Lustre 的定制网络层，支持 InfiniBand、RoCE、TCP/IP 等多种传输 AI 场景配置建议：

Requires Lustre client 2.15+

Create training data directory

lfs setstripe -c 16 -S 4M /lustre/training-data/

Create checkpoint directory (medium striping)

lfs setstripe -c 8 -S 1M /lustre/checkpoints/

View file striping info

lfs getstripe /lustre/training-data/model-weights.dat 性能调优关键参数： • max_dirty_mb ：客户端脏数据上限，影响写入缓冲行为 • max_rpcs_in_flight ：并发 RPC 数，增加此值可提高带宽但增加延迟 •使用 InfiniBand/RoCE 时启用 LNet Multi-Rail 以同时使用多个网络接口 优点：成熟稳定（20+ 年发展），社区庞大，HPC 生态无缝集成，聚合带宽可达 TB/s。缺点：元数据性能是瓶颈（单 MDS），InfiniBand 依赖，运维复杂（内核模块偶需调试）。 2) GPFS GPFS（现称 IBM Spectrum Scale）是 Lustre 在高端企业市场的主要竞争对手，以功能完整性和强大的数据管理著称。 架构特色： •NSD（Network Shared Disk）：类似 Lustre OST 的存储块设备，可直接通过 SAN 光纤网络访问 •分布式字节范围锁（Distributed Byte-Range Locking）：字节级锁粒度，允许同一文件的多个写入者同时操作不同字 节范围，这是集群文件系统的难题 •AFM（Active File Management）：文件数据在站点间异步复制，适合多数据中心 AI 训练 GPFS 使用分布式元数据，多个节点可同时处理元数据请求，解决了 Lustre 单 MDS 的瓶颈问题。对于包含数十亿小文件 的 AI 数据集，分布式元数据是决定性的性能优势。 AI 适用性：GPFS 在 IBM Watson 和 IBM Research 的 AI 集群中得到验证，支持 GPUDirect Storage（GPU 绕过 CPU，通 过 RDMA 直接从 NSD 读取数据）。许可证成本是阻碍广泛采用的主要因素。 3) WekaFS WekaFS 专为 AI/ML 和 HPC 云原生场景设计，核心特点包括： •分层存储（Tiered Storage）：将热数据保持在 NVMe SSD，冷数据自动迁移到对象存储（S3） •无状态客户端：客户端不需要任何持久化状态，节点故障不影响数据完整性（与 Lustre 的有状态客户端形成对比） •DPU 卸载：在 NVIDIA BlueField DPU 上运行 WekaFS 服务，释放 CPU 资源 •Cloud-Native Snapshots：支持 snapshot 快速克隆，适合 ML 实验分支 WekaFS 的并行性来自分布式 hash 表（DHT）架构，每块数据通过一致性 hash 定位到特定 NVMe 驱动器，避免集中式 元数据服务器的瓶颈。 基准数据（Weka 官方，2024）： •单客户端吞吐：最高 113 GB/s（使用 8× NVIDIA CX-7 400Gbps 网卡） •元数据性能：最高 2.5M IOPS（文件创建） •可扩展至 1000+ 客户端节点 虽然是纯软件方案，但 WekaFS 的许可证费用是 Lustre 社区版的数倍。对预算充足的云原生 AI 部署，WekaFS 提供了最 少的运维工作量。 4) BeeGFS BeeGFS 起源于德国 Fraunhofer 研究所，以简单、高效著称，架构极简： •管理服务：单点，轻量（仅需管理配置） •元数据服务：分布式，文件/目录元数据可按目录分片 •存储服务：每个存储服务器运行用户态进程，通过 TCP/RDMA 暴露存储 •客户端：用户态或内核态均可，通过简单的 POSIX 接口挂载 BeeGFS 的全用户空间设计避免了内核模块的复杂性，部署和调试门槛极低。在 TOP500 列表中支持了多套入门级 HPC 系统，AI 场景下适合对运维成本敏感的中小型团队。 5) 3FS DeepSeek 的 3FS（Fast File System，又称 Fire-Flyer File System）是近年来 AI 领域最受关注的定制存储系统之一。核 心设计理念： •完全分离式架构：数据平面在存储节点，控制平面在独立元数据集群 •RDMA 直通：所有数据路径均通过 RDMA 单边操作（READ/WRITE），零 CPU 复制 •链式复制（Chain Replication）：写请求依次通过 Primary → Secondary → Tertiary 节点，在保证一致性的前提下最 大化写入吞吐 •无中心元数据：元数据分片到多个元数据节点，通过 CRAQ 协议保证强一致性 3FS 的设计直接回应了 LLM 训练的痛点，即检查点写入的爆发性高吞吐需求。通过 RDMA 直通和链式复制，3FS 可在数 千节点上实现 TB/s 级的聚合写入带宽（生产环境实测约 6.6 TiB/s 聚合读），将 175B 模型的检查点写入时间压缩到 30 秒 以内。DeepSeek-V2（236B MoE）和 V3（671B MoE）的训练均基于 3FS。 6) Ceph 与 JuiceFS Ceph 的强项在统一，一套集群同时提供对象（RGW）、块（RBD）与文件（CephFS）三种接口，元数据由 MDS 集群按 动态子树分区承载，并依托 CRUSH 放置与 PG 自愈实现免人工运维。AI 场景的取舍：追求极致带宽与超大目录选 Lustre；需要一套存储同时服务 S3 数据湖、块设备与 POSIX 访问、且规模在 PB 以内，Ceph 更经济。Ceph 的复杂性主 要来自 PG 数量调优与亚秒级延迟抖动，超大规模（10+ PB）部署仍以 Lustre/GPFS 为稳妥选项。 JuiceFS 是对象存储之上的分布式文件系统，元数据放入独立元数据引擎（Redis/MySQL/TiKV 等），数据本体存对象存 储，天然实现分层。其容量随对象存储近似无限，适合云上训练数据与检查点归档、多集群共享等成本敏感场景；高并发 随机读需依赖本地缓存兜底。 各系统综合对比如表13-3所示。 表13-3 并行文件系统综合对比 系统 架构 元数据 POSIX 语义RDMA 扩展性 运维 许可 适用场景 支持 难度 证 Lust 有状态客户 集中式 完整 POSIX， InfiniBa nd/RoC 数据面 TB/s 级，单 MDS 高 开源 GPLv 千卡以上训练的 数据集与检查点 re 端 MDS/MDT 强一致 E 元数据瓶颈 2 热层 GPF 有状态客户 分布式元数 据，字节范围 完整 POSIX， InfiniBa nd/RoC 分布式元数据支撑十亿级 小文件，AFM 跨数据中 中 商业 企业多云训练， S 端 锁 强一致 E 心复制 许可 GDS 原生支持 Wek 无状态客户 分布式 DHT 完整 POSIX InfiniBa nd/RoC TB/s 级，可扩展至 低 商业 云原生 AI 训练， aFS 端 E 1000+ 客户端 许可 运维最简 Bee 全用户态 按目录分片 完整 POSIX InfiniBa nd/RoC 100+ GB/s 低 开源 GPLv 成本敏感的中小 GFS E 2 团队 分离式 部分 POSIX， Native TB/s 级（实测读约 6.6 超大规模训练的 3FS RDMA 分片 + CRAQ 面向顺序大文 RDMA TiB/s） 中 私有 检查点与数据集 件 Cep 无状态客户 MDS 集群动态 完整 （内 POSIX msgr2/ PB 以内经济，10+ PB 部 高 开源 一套存储同时服 hFS 端 子树分区 核/FUSE） RDMA 署复杂 LGPL 务对象/块/文件 Juic 对象存储之 独立元数据引 完整 POSIX 底层走 对象存 容量随对象存储近无限 中 开源 云上训练与归 eFS 上的文件系 统 擎 （Redis/TiKV） （FUSE） 储 AGPL 档，成本敏感 选型结论应落到三个数字上：目标聚合带宽、文件数与元数据 IOPS、可接受的单 GB 成本。对于大多数 AI 团队，Lustre 是经过验证的默认选择；云原生部署场景 WekaFS 或 AWS FSx for Lustre 提供更低的运维负担；DeepSeek 3FS 的架构思 想（RDMA 直通 + 链式复制）对未来定制 AI 存储方案具有重要的参考价值。 GDS 与 RDMA 存储加速 GPU Direct Storage（GDS）是 NVIDIA 从 CUDA 11.4 起引入的关键存储加速技术，使 GPU 能通过 RDMA 直接从 NVMe SSD 或并行文件系统读取数据到 GPU HBM，完全旁路 CPU 内存。传统路径为 NVMe → CPU DRAM（DMA）→ GPU HBM（PCIe），内存带宽消耗加倍；GDS 路径为 NVMe → GPU HBM（单次 DMA），数据加载带宽提升 30-50%，同时释 放 CPU 内存带宽。在 AI 训练场景中，GDS 可将每个 batch 的加载延迟从 50-200 ms 降低到 30-120 ms，在 1000 GPU 集群上相当于回收 2-5% 的训练吞吐；在推理场景中可加速模型加载，冷启动时间缩短 40-60%。 GDS 对文件系统有格式要求：需以 O_DIRECT 模式打开文件（旁路 OS page cache），使用兼容的本地文件系统（如 ext4/xfs on NVMe 普通挂载），或通过 MOFED 的 nvme-cli 配置 NVMe-oF target。在并行文件系统层面， GPFS/Spectrum Scale 原生支持 GDS；Lustre 从 2.15 起通过 nvidia-fs.ko 对其 O_DIRECT 路径的适配提供实验性 GDS 支持，应用层使用标准 cuFile API 访问；WekaFS 通过其无状态客户端架构天然兼容 GDS。

13.1.3 对象存储与数据湖

对象存储（Object Storage）传统上被视为冷数据和归档存储的廉价方案，但在大模型时代，对象存储正在向 AI 工作负 载的前端移动，从“数据湖的后端”进阶为“AI 数据管线的第一公民”。S3 兼容 API 已成为 AI 数据生态的通用接口。 •S3 与 AI 负载：对象存储的核心特性与 AI 场景的契合点： •海量小文件：AI 数据集通常包含数亿到数十亿个小文件（图像、文本片段）。对象存储的扁平命名空间天然支持这种 规模，S3 单桶可容纳无限数量的对象，按 key 的单对象访问为常数级定位，无需目录线性扫描 •不可变性：训练数据是 write-once-read-many 的典型场景，与对象存储的模型完美匹配 •多协议访问：同一数据可通过 S3 REST API 被预处理流水线访问，也可通过 POSIX 网关被训练代码访问 •生命周期管理：自动将热数据保持在 NVMe，冷数据迁移到 HDD/磁带 以 ImageNet（150 万张图片，约 150 GB）为例，对象存储按 key 的单对象访问可在毫秒级完成；但列出全部对象需 要分页迭代（ListObjectsV2 每页上限 1000 个对象，共需上千次请求）。相比之下，POSIX 文件系统在包含百万文件的 单目录下， ls 操作可能需要数十秒甚至分钟。 •MinIO：MinIO 是 AI/ML 社区使用最广泛的开源 S3 兼容对象存储，其设计理念是“高性能、简单、云原生”。 关键架构特性： •每个 MinIO 服务器实例是平等的（无 master/worker 之分），任何节点可处理任何请求 •使用 Reed-Solomon Erasure Coding（默认 EC:4，即 50% 冗余开销），容忍多达半数节点失效 •不支持 POSIX FUSE 挂载（MinIO 的官方立场是：POSIX 不适合云原生对象存储） AI 场景下的 MinIO 配置示例： # docker-compose.yml with 4-node MinIO cluster version: "3.8" services: minio1: image: minio/minio:RELEASE.2024-04-18T19-09-19Z command: server --console-address ":9001" http://minio{1...4}/data{1...4} environment: MINIO_ROOT_USER: admin MINIO_ROOT_PASSWORD: minio123 volumes: - /nvme0/minio1:/data1 - /nvme1/minio1:/data2 network_mode: host 性能基准（4 节点 NVMe，100 GbE 网络）：

• GET 吞吐：约 40 GB/s（64 并发对象读取）
• PUT 吞吐：约 30 GB/s（64 并发对象写入）
• LIST 操作：约 10K objects/sec MinIO 在 PyTorch 中可直接作为 S3Dataset 的数据源（通过 s3fs 库或 MinIO 的原生 Python SDK） ：

    # Requires Python 3.10+ and boto3
    import boto3
    from io import BytesIO
    from PIL import Image
    s3 = boto3.client('s3', endpoint_url='http://minio-server:9000',
                      aws_access_key_id='admin', aws_secret_access_key='minio123')
    def load_image_from_s3(bucket, key):
        obj = s3.get_object(Bucket=bucket, Key=key)
        return Image.open(BytesIO(obj['Body'].read()))

•Ceph RGW：Ceph RADOS Gateway（RGW）提供 S3 兼容接口，同时支持块存储（RBD）和文件存储（CephFS）。 Ceph 的统一架构优势在于，一个 Ceph 集群可同时服务： •训练数据的 S3 访问（通过 RGW） •检查点写入的高性能 POSIX（通过 CephFS） •GPU 节点操作系统的块设备（通过 RBD） Ceph 的复杂性使其在超大规模（10+ PB）AI 部署中较少被选为首选方案。其 CRUSH 算法数据放置策略和 PG （Placement Group）自治修复模型在亚 PB 级别可能引入不可预测的延迟抖动。 •POSIX 网关：将对象存储暴露为 POSIX 文件系统的工具包括： •s3fs-fuse：最成熟的 S3 FUSE 实现，每个文件操作转换为 S3 API 调用 •Mountpoint-S3（AWS 官方）：基于 Rust 重写的高性能 S3 FUSE，专为大文件顺序读取优化，延迟比 s3fs 低 10-50 倍 •Goofys：更轻量的 S3 FUSE，牺牲部分 POSIX 兼容性换取高性能 对于 AI 训练场景，Mountpoint-S3 的优势显著：

    # Requires AWS Mountpoint-S3 client
    # Mount S3 bucket as a local filesystem
    mount-s3 --prefix training-data/ my-bucket /mnt/training
    # PyTorch DataLoader can read directly
    dataset = ImageFolder("/mnt/training/imagenet/train")

关键限制：POSIX 网关无法将 S3 的最终一致性模型变为严格的 POSIX 一致性。对于检查点写入等强一致性需求，直接使用原生并行文件系统或对象存储的原 生 API 更为可靠。 •AI Data Lake：现代化的 AI Data Lake 通常将数据以列式格式（Parquet、ORC）存储在对象存储上，实现以下收益： •Predicate Pushdown：查询引擎（Spark、DuckDB）将过滤条件下推到存储层，减少数据传输 •Column Pruning：仅读取所需的列（在 tokenization 中可能只需 text 列，无需 metadata 列） •压缩效率：列式格式的压缩比通常为 2-5×（Parquet + Snappy/ZSTD） 典型 AI Data Lake 目录结构： s3://ai-dataset/ ├── raw/common-crawl/wet/2024-01/... ├── processed/cc-filtered/parquet/year=2024/month=01/ ├── tokenized/llama-tokenizer/seqlen=8192/ └── models/llama-70b/checkpoints/step=1000/ 使用 Apache Iceberg 或 Delta Lake 表格格式管理 ML 数据集，可提供 ACID 事务、时间旅行和 schema 演化等数据仓库级能力。 常见存储方案的硬件成本、运维成本与延迟范围对比如表13-4所示。 表13-4 存储方案成本对比 存储方案 硬件成本 (per TB) 年运维成本 (per PB) 大数据集规模 延迟范围 全闪存 Lustre $2,000-3,000 $50K-100K < 1 ms 混合 Lustre (SSD+HDD) $500-1,000 $30K-50K PB级 1-100 ms WekaFS (on-prem) $1,500-2,500 $40K-80K (含授权) PB级 < 1 ms MinIO (commodity HW) $200-400 $10K-20K 数十 PB 1-50 ms 云对象存储 (S3 Standard) — $23/TB-month 无限制 10-100 ms 云 FSx for Lustre — $0.09/GB-month PB级 < 1 ms 云 EFS (AWS) — $0.30/GB-month 无限制 1-5 ms 对于频繁访问的训练数据和检查点，并行文件系统（Lustre/GPFS/WekaFS）的低延迟和高带宽价值巨大；对于基础数据 集、日志长期归档、多区域分发，对象存储是更经济的选择。典型的混合方案是并行文件系统承载热数据和活跃工作区， 对象存储作为数据湖和冷备份。

13.1.4 元数据与小文件问题

AI 数据集常包含数十亿小文件（如图像分类的独立 JPEG、语音识别的独立 WAV），使元数据操作成为实际瓶颈。在 POSIX 文件系统上，单目录含 100 万文件时 ls 需 30-90 秒， find 遍历可能耗时数分钟，因为 POSIX 目录按线性列表 存储 dirent。并行文件系统对此缓解有限：Lustre 的 DNE 将不同目录分布到不同 MDT，但单目录内仍是线性；GPFS 的 分布式元数据可并行处理单目录请求，对小文件场景优势显著。 针对此问题，工程实践沉淀出五类缓解策略： •Packing 打包：将小文件打包为大文件（如 WebDataset 的 tar shard 格式、TFRecord），降低 inode 数量与元数据请 求次数，代价是失去随机访问能力 •目录扁平化：按 shard 或日期分目录（如 tokenized/seqlen=8192/shard-00000.pt ），把文件摊开，避免单目录上 万文件时的线性遍历 •对象存储绕行：利用对象存储的扁平命名空间，按 key 的单对象访问无需目录线性扫描（LIST 仍需分页迭代，每页上 限 1000 个对象），海量小文件归档放对象存储，POSIX 层只保留活跃子集 •元数据与数据分离：3FS 的分片元数据集群、Ceph 的分布式 MDS、JuiceFS 的独立元数据引擎缓解了集中式瓶颈，代 价是系统复杂度上升 •内容寻址格式：如 DVC/Pachyderm 的内容寻址去重，按内容哈希去重存储 对于训练场景，建议预处理阶段保留数据集索引文件（JSONL manifest），训练时通过索引直接定位到分片和偏移量，避 免文件系统级别的元数据扫描。

13.2 检查点存储设计

大模型训练的检查点（Checkpoint）写入是 AI 存储系统面临的最严峻挑战。以 Llama-70B 训练为例，完整检查点包含 模型参数、优化器状态（Adam 的 momentum 和 variance）、学习率调度器状态和随机数种子，按混合精度全量口径 （16 字节/参）总大小约 1.1 TB。在千卡集群上，每 N 步（通常 100-1000 步）需保存一次，要求存储系统提供几十 GB/s 乃至 TB/s 的聚合写入带宽。

13.2.1 检查点 I/O 特征与分片保存

检查点写入的独特特征： •爆发性：所有 GPU 几乎同时开始写入，压力集中在 30-120 秒时间窗口 •写吞吐主导：很少读取（仅在故障恢复时），写:读比例 > 100:1 •大块顺序写：每 GPU 写一个（或少数几个）大文件块，I/O 大小通常 100 MB-2 GB •严格一致性：所有 GPU 的检查点分片必须属于同一训练步，任何不一致都会导致恢复失败 对于 175B 模型（全量约 2.8 TB），1000 卡集群的检查点窗口为 60 秒，需要的聚合写入带宽约 47 GB/s，规划时按 ≥50 GB/s 预留以容纳网络和软件开销。 分片保存（Sharded Save）是指每 GPU 仅保存其持有的模型分片和优化器状态分片，这是 FSDP 和 DeepSpeed ZeRO 的标准策略。

DeepSpeed ZeRO-3 checkpoint save

ds_config = { "zero_optimization": { "stage": 3, }, "checkpoint": { "use_node_local_storage": True, # Use node-local NVMe as intermediate cache "max_io_workers": 8, # Number of write threads per rank } } 异步检查点需要 CPU 侧缓冲暂存模型状态，ZeRO-3 分片下每 GPU 缓冲约为模型大小的 1/N（70B 模型千卡规模每 GPU 约 140 MB），开销可忽略。

13.2.2 多层架构与异步缓冲

主流 LLM 训练框架采用“内存 → NVMe → 并行文件系统 → 对象存储”的分级检查点策略，如图13-2所示。 GPU HBM Model State torch.save async Node-Local NVMe SSD Buffer Layer Background Sync Parallel File System Lustre/GPFS Async Archive Object Store S3/MinIO Disaster Recovery Remote Backup Cross-Region S3 图13-2 多层检查点存储架构 •L1 GPU HBM：训练状态在 GPU 显存中，延迟 < 1 μs •L2 节点本地 NVMe：提供 3-7 GB/s 的单节点写入带宽，将全局检查点时间完全与存储系统解耦。NVMe 上的检查点可 保留最近 K 个（如 K=3），作为“快速回滚”的缓存 •L3 并行文件系统：作为权威检查点存储，提供全局命名空间，所有节点共享 •L4 对象存储：长期归档，成本最低 节点本地 NVMe 层缓冲了写爆发。即使 Lustre 的聚合带宽未达到理想水平，只要 NVMe 缓冲足够大（如 4 TB × 8 节点 = 32 TB），检查点就能在数秒内完成 GPU → NVMe 的阶段（L1 → L2），后续 L2 → L3 的后台同步可从容进行。

13.2.3 检查点压缩

检查点数据对压缩算法友好。FP16 参数矩阵和 FP32 优化器状态的熵通常为 6-10 bit/参数（即 30-50% 的冗余）。常见压 缩策略： •ZSTD 压缩：Level 3-5 提供约 2× 压缩比，吞吐可达 1-3 GB/s per core。建议在检查点写入前对每个分片执行 ZSTD 压缩 •量化存储：将检查点的参数从 FP32 量化为 FP16/BF16 存储，代价是恢复时需重新展开。DeepSpeed 的 bf16 checkpoint 选项（ "checkpoint_parallel_write_pipeline": True ）即采用此策略 •分块压缩：每个 GPU 独立压缩其分片，并行度 100%

13.2.4 检查点直达对象存储

直接将检查点写入 S3 兼容对象存储（绕过并行文件系统）是新兴实践。AWS 的 S3 Express One Zone（单 AZ，亚毫秒 级延迟）和 MinIO 的高性能 PUT 路径使其成为可能。优势包括： •跳过昂贵的中层并行文件系统 •原生的生命周期管理（自动从 S3 Standard → S3 Glacier） •原生多区域复制（Cross-Region Replication）用于灾难恢复 S3 直接 Checkpoint 配置（PyTorch + boto3）：

 # Requires Python 3.10+, PyTorch 2.x and boto3
 import boto3, torch, io
 s3 = boto3.client('s3')
 buffer = io.BytesIO()
 torch.save({"model": model.state_dict(), "optimizer": optimizer.state_dict()}, buffer)
 buffer.seek(0)
 s3.upload_fileobj(buffer, "ai-checkpoints", "step-1000/model.pt")

权衡：对象存储的延迟通常为 10-50 ms（S3 Standard），大规模并发 PUT 可能遇到 S3 的速率限制（默认 3500 PUT/sec per prefix），需要仔细设计分片命名避免 prefix hotspotting。

13.2.5 高速写路径与存储选型

checkpoint 读写吞吐直接关系 GPU 利用率。写入窗口内训练要么同步阻塞，要么依赖异步路径暂存，窗口越长 GPU 空 转越久；恢复时整集群空闲等待数据回读，吞吐同样关键。以 Llama-70B 全量检查点约 2 TB 为例，20 秒内写完需要约 100 GB/s 聚合写带宽（2 TB / 20 s），放宽到 60 秒则约 33 GB/s。相对训练进度，写入气泡占比约为 T /(N × T )， 其中 T 为写入窗口，N 为两次 checkpoint 间的步数，T 为单步时长。 ckpt step ckpt step checkpoint 的写路径沿 GPU 显存 → 节点本地 NVMe → 并行文件系统（权威副本）→ 对象存储（归档）四层展开。节点 本地 NVMe 缓冲把阻塞训练的阶段压缩为 GPU → 本地盘一段，后台同步与训练步并行推进。同步直写路径上，GPU 需 等待检查点全部落盘才继续训练，写入窗口直接转化为训练气泡；异步路径先将状态写入节点本地 NVMe，再后台同步到 并行文件系统，训练几乎不受影响。 存储侧需要从三个维度支撑 checkpoint： •突发吞吐：checkpoint 是短时突发写，聚合带宽按窗口内写完全量预留，写入摊开到足够多的 OST 与 OSS 上，避免 单点热点 •容量：并行文件系统保留最近 K 个（如 K=3）checkpoint 供快速回滚，历史版本异步归档到对象存储，仅消耗归档成 本 •元数据：单步 checkpoint 在千卡规模下产生约 1024 个文件，训练累积使检查点目录膨胀到百万级文件，元数据服务 器压力集中在目录项创建与 stat 操作 训练数据、checkpoint 与推理模型仓库三类数据对存储的要求不同，不应共用一个方案，选型对比如表13-5所示。 表13-5 训练三类数据存储选型 数据 访问特征 带宽要求 存储选型 训练数据 顺序/伪随机读，长期复用 数百 MB/s 到数 GB/s 聚 对象存储底座 + 本地 NVMe 缓存，或并行文件系统 合 热层 checkpoint 突发大块写，读少但恢复时全量 读 数十到数百 GB/s 峰值 并行文件系统 + 节点本地 NVMe 缓冲 推理模型仓 启动时一次性大块顺序读 单节点 1-10 GB/s 对象存储为主，配分布式拉取 库 本地缓存加速是把带宽需求从存储网络转移到节点内的关键手段。训练数据以对象存储为底座，把热点 shard 缓存到节 点本地 NVMe（每节点约 15 TB 的规模），DataLoader 命中缓存即绕过存储网络，未命中再回源对象存储。相比全量读取 并行文件系统，对象存储 + 本地缓存在带宽、成本与可扩展性之间更均衡，适合云上或自建 S3 兼容环境。

13.2.6 检查点工程清单

检查点存储的工程决策汇总如表13-6所示。 表13-6 检查点工程清单 需求 推荐方案 关键参数 减少训练停顿 节点 NVMe 缓冲 L1→L2 本地暂存 + 后台同步 减少存储容量 ZSTD 压缩 + 保留最近 K 个 ZSTD level 3, K=3 灾难恢复 跨 AZ 复制到对象存储 CRR / S3 复制规则 长时间训练 存储级去重快照 ZFS snapshot / Lustre HSM 超大规模 (>100B) 分片写入 + S3 直达 每节点独立压缩 + 并发 PUT 检查点存储的设计对训练工程的可靠性有决定性影响。当训练运行在 1000+ GPU 上数周时，检查点的可靠性和恢复速度 直接决定了训练的成功率。

13.3 向量数据库与版本管理

向量数据库（Vector Database）是支撑 RAG（检索增强生成）、语义搜索、推荐系统等 AI 应用的核心存储组件，其本质 是以向量嵌入（Embedding）为索引、实现近似最近邻（ANN, Approximate Nearest Neighbor）搜索的专用存储系 统。在大模型时代，向量数据库不仅是存储组件，更是连接模型与知识的“语义记忆层”。本节同时覆盖数据版本管理与 特征存储，它们是保障 AI 系统可复现性的数据工程基础设施。

13.3.1 向量索引基础

精确的最近邻搜索（KNN）在十亿级向量上的复杂度为 O(N ⋅ D)（N 为向量数，D 为维度），完全不可行。ANN 将精度 要求从“精确”放宽到“近似”，换取了数量级的速度提升。 核心概念： •召回率（Recall@K）：ANN 返回的 top-K 中包含真实 top-K 的比例，典型目标为建设 95-99% •QPS（Queries Per Second）：每秒处理查询数 •构建时间（Build Time）：全量向量索引的构建耗时 •内存占用：索引结构的存储开销 ANN 的核心洞察是“维度灾难”的另一面：高维空间稀疏，大多数数据点彼此远离，仅极少数邻点需要精确确定。这使 得粗筛选 + 精排的二阶段策略成为可能。 主流索引类型：

1. IVF •使用 K-Means 将所有向量聚类为 n 个簇 list •查询时先与聚类中心计算距离，选择最近的 n 个簇，仅在这些簇内进行精确距离计算 probe •时间复杂度 O(n ⋅ ⋅ D) N probe •召回率取决于 n （一般 n ≈ N , n = 8 − 32） nlist probe list probe
2. HNSW •构建多层图结构：顶层稀疏（长距离连接），底层稠密（短距离连接） •查询从顶层开始贪心遍历，逐层下降到底层搜索最近邻 •时间/空间复杂度 O(log N ⋅ M ⋅ D)（M 为每节点的连接数） •索引构建较慢，但查询速度最优（最常用的 ANN 索引算法）
3. PQ •将 D 维空间分解为 M 个子空间（每个 维），每个子空间有独立的聚类中心码本 D •向量被压缩为 M 个码本索引的组合（如 M = 8，每子空间 256 个聚类中心，则向量用 8 bytes 表示） M •查询时距离通过查询码本预计算的距离表进行查表求和 •压缩效率极高（D=768 → 8 bytes，压缩比 384×），但召回率稍低（需配合 IVF 或 HNSW 使用）
4. DiskANN •Microsoft 专为 SSD 设计的 ANN 索引，核心创新是将图索引结构和向量数据分开存储 •SSD 上的图遍历通过预取（prefetching）和批量读取来规避随机读延迟 •十亿级向量可在 64 GB DRAM + 1 TB SSD 上实现 < 10 ms 的查询延迟 •是成本敏感场景的优秀选择

13.3.2 向量库存储架构

从存储视角看，主流向量数据库的差异集中在持久化方式与索引存储布局： •持久化方式：pgvector 嵌入 PostgreSQL 生态，复用其事务语义，将向量与关系数据统一存储；Milvus 采用分布式架 构，支撑十亿级向量规模；Pinecone 为托管 Serverless 服务，零运维，但数据须上传云端，对数据主权敏感的应用需 谨慎 •索引存储布局：Qdrant 以 HNSW 为主索引，通过 Scalar Quantization（INT8 压缩，4× 内存节省）和 Product Quantization 降低内存占用；pgvector 提供内存中的 IVFFlat（基于 K-Means 聚类，与 FAISS IVF 等价）和 HNSW （PG 14+）两种索引，十亿级规模依赖 PostgreSQL 分区表与并行查询水平扩展 •版本管理与备份恢复：向量数据的版本管理与备份恢复在存储层实现，依托分段与对象存储归档机制，是部署架构设计 的一部分 向量索引的内存布局直接影响查询性能。常见内存布局形式： •行存（Row-wise）：[v [0], v [1], …, v [D − 1], v [0], …]，相邻向量在内存中连续。适合点查询（获取单个向量），但 1 1 1 2 ANN 搜索中需要计算查询向量与多个数据向量某维度的距离，行存导致频繁的跨缓存行访问 •列存（Column-wise）：[v [0], v [0], …, v [0], v [1], …]，每列包含所有向量的同一维度值。对 ANN 距离计算友好， 1 2 1 SIMD 可批量计算所有向量与查询向量在某一维度的差值，利用 L1/L2 缓存行 N 对于 DiskANN 等磁盘索引，列存几乎必须，因为从 SSD 读取的每个页（4 KB）都应包含最大量的计算相关信息。 向量数据库在 RAG pipeline 中的位置如图13-3所示。 Embedding Model Vector Database (Document Chunk Vectors Retrieved Top-K Documen LLM Generation User Query text-embedding-3-large ANN Search + Metadata) t Chunks [Context + Query] -> Answ User Answer er 图13-3 RAG中的向量数据库工作流 典型配置参数（基于 Cohere 或 OpenAI Embedding，维度 1024/1536）： •索引类型：HNSW（高召回，查询快） •M （连接数）：16-32（越高召回越好但内存越多） •ef ：200（构建时的搜索深度） construction •ef ：64-128（查询时搜索深度，平衡延迟和召回率） search •向量压缩：不压缩（保持全精度）或 PQ 压缩（损失 1-3% 召回率） 查询延迟（100 万向量，HNSW, D=1536, NVMe SSD）： •无量化：p50 = 2 ms, p99 = 8 ms •PQ 压缩（8×）：p50 = 1 ms, p99 = 5 ms 对于 RAG 场景，延迟控制在 10 ms 以内（embedding 提取约 20 ms + 向量搜索约 5 ms + LLM 生成约 500 ms = 总计约 525 ms）是可接受的，向量搜索远非系统的瓶颈。

13.3.3 十亿级部署分层存储

┌──── Hot segment (last 30 days) ────┐ Client ──▶ LB ──▶ Query Nodes ──▶ In-memory index (HNSW) │ ┌──── Warm segment (30-365 days) ────┐ └──▶ NVMe SSD index (DiskANN) │ ┌──── Cold segment (>365 days) ──────┐ └──▶ Object storage (S3/MinIO) archive 分段策略： •时间分区：按文档的创建/更新时间分段，逐段独立索引 •热度分层：高频查询段驻留内存（HNSW），中频段用 DiskANN（SSD），低频段归档到 S3（查询时加载到临时缓存） •动态迁移：根据查询模式自动调整分段热度层级 这种架构的 TCO 比全内存方案低 60-80%（多数索引数据在查询中从未被命中），延迟代价仅在冷段命中时增加 50-200 ms（S3 加载延迟）。

13.3.4 数据版本管理

AI 系统的可复现性不仅依赖模型代码和超参数，更依赖数据的精确版本。在 ML 工程实践中，“能复现上周的模型结果” 是最常见也是最困难的需求之一。数据集版本管理（Data Versioning）和特征存储（Feature Store）是该问题的两个核 心解决范式。 •数据版本管理挑战：与传统软件工程的代码版本管理（Git）不同，ML 数据版本管理面临独特挑战：

1. 数据量：MB/GB 级代码 vs TB/PB 级数据集
2. 格式多样性：图像、文本、视频、表格、嵌入向量
3. 派生关系：原始数据 → 清洗 → 标注 → 特征工程 → 训练集/验证集/测试集分裂，每个步骤都可能产生新版本
4. 部分更新：新增 1000 条标注数据，修改 50 条错误标签，不应触发全量数据复制 理想的数据版本管理方案应当支持类似 Git 的语义（commit, branch, diff, checkout），但操作对象是 PB 级数据集。 •DVC 数据版本控制：DVC（Data Version Control）是最成熟的开源数据版本管理工具，采用“代码在 Git，数据在对 象存储”的分离架构：
5. 数据集被哈希计算（MD5），生成 .dvc 元数据文件（类似指针/引用）
6. .dvc 文件被提交到 Git，实际数据存储在远程存储后端（S3, GCS, Azure Blob, SSH, HDFS）
7. dvc push/pull 在 Git 的工作流上增加了数据同步

 # Requires DVC 3.x
 # DVC typical workflow
 # 1. Initialize

dvc init git commit -m "Initialize DVC"

2. Start tracking dataset

dvc add data/raw/training-images/ git add data/raw/training-images.dvc data/.gitignore git commit -m "Add raw training images v1"

3. Configure remote storage

dvc remote add -d myremote s3://my-bucket/dvc-store/ git commit -m "Configure remote storage"

4. Push data to remote

dvc push

5. After dataset update

dvc add data/raw/training-images/ # Re-track git commit -m "Update training images v2" dvc push

6. Revert to old data

git checkout v1.0 # Switch to old commit dvc checkout # Sync data to that version DVC 的 dvc run 或 dvc stage add 可定义数据处理 DAG： # dvc.yaml - Data processing pipeline definition stages: preprocess: cmd: python preprocess.py data/raw data/processed deps: - data/raw - preprocess.py outs: - data/processed train: cmd: python train.py data/processed model.pkl deps: - data/processed - train.py outs: - model.pkl metrics: - metrics.json: cache: false DVC 自动追踪每个阶段的输入/输出依赖关系，执行 dvc repro 时仅重新运行依赖变更的阶段。 •Delta Lake 与 LakeFS：Delta Lake（Databricks）将 ACID 事务引入 Data Lake，通过 Parquet + 事务日志实现数据 版本管理：

 # Requires PySpark 3.x and delta-spark
 from delta.tables import DeltaTable
 # Write to Delta table
 df.write.format("delta").mode("overwrite").save("/delta/training-data")
 # Time travel
 df_v1 = spark.read.format("delta") \

.option("versionAsOf", 0)
.load("/delta/training-data")

 # View version history
 deltaTable = DeltaTable.forPath(spark, "/delta/training-data")
 deltaTable.history().show()

Delta Lake 的数据版本通过事务日志（ _delta_log 目录）中的 JSON/Parquet 检查点文件实现：写入时追加新日志条 目，读取时从日志重放构建最新状态，每次写入产生一个单调递增的“版本号”。 LakeFS（Treeverse）从另一个角度解决数据版本问题，不是数据库层的事务，而是对象存储之上的“Git-like 分支层”。 LakeFS 在对象存储（S3/MinIO）之上创建 branch/tag/commit 语义，数据操作在隔离分支上进行，merge 时声明性地 选择最新版本或手工解决冲突。 # Requires lakeFS CLI (lakectl) lakectl branch create lakefs://myrepo/experiment-1 -s lakefs://myrepo/main lakectl fs upload -r lakefs://myrepo/experiment-1/train/ ./new-data/ lakectl commit lakefs://myrepo/experiment-1 -m "Add new training data" lakectl merge lakefs://myrepo/experiment-1 lakefs://myrepo/main LakeFS 的独特优势：多个 ML 实验可以并行运行在不同的数据分支上，互不干扰（类似 Git 的 feature branch）。

13.3.5 特征存储与数据血缘

•特征存储：特征存储解决的是训练-推理的一致性鸿沟（Training-Serving Skew），其核心价值主张是训练时使用的特 征转换逻辑应与推理时完全一致，且特征应被复用（而非每个模型团队各自从头计算）。 Feast（Gojek 开源，2019，现为 LF AI & Data Foundation 托管项目）是最广泛使用的开源特征存储：

   # Requires feast 0.30+
   from feast import Entity, FeatureView, Field, FileSource
   from feast.types import Float32, Int64
   driver = Entity(name="driver_id", join_keys=["driver_id"])
   driver_stats_source = FileSource(
       path="s3://my-bucket/driver_stats.parquet",
       timestamp_field="event_timestamp",

)

   driver_stats_fv = FeatureView(
       name="driver_hourly_stats",
       entities=[driver],
       ttl=timedelta(days=7),
       schema=[

Field(name="conv_rate", dtype=Float32), Field(name="acc_rate", dtype=Float32), Field(name="avg_daily_trips", dtype=Int64), ], source=driver_stats_source, )

   # Training-time feature retrieval
   from feast import FeatureStore
   store = FeatureStore(repo_path=".")
   training_df = store.get_historical_features(
       entity_df=orders_df,
       features=["driver_hourly_stats:conv_rate", "driver_hourly_stats:acc_rate"],

).to_df()

   # Inference-time feature retrieval
   feature_vector = store.get_online_features(
       entity_rows=[{"driver_id": 1001}],
       features=["driver_hourly_stats:conv_rate", "driver_hourly_stats:acc_rate"],

).to_dict() 在线与离线特征服务：

• 离线（Offline）：支持大时间范围的批量特征提取，使用 Parquet/Delta Lake 等列式格式，PB 级
• 在线（Online）：低延迟（< 10 ms）特征检索，使用 Redis/Cassandra/DynamoDB 等 KV 存储 核心挑战：离线批量计算的特征值与在线实时计算的特征值必须一致。不一致的根源包括：

1. 聚合窗口边界不同（离线按自然日，在线按滚动窗口）
2. 空值处理逻辑不同（离线填 0，在线填均值）
3. 特征依赖的时间对齐，一个请求中的多个特征来自不同的时间点 Tecton 是 Feast 的商业化版本，增加了特征编排、特征监控、自动特征工程等企业级能力，但保持了 Feast 核心 API。 •数据血缘追踪：数据版本管理和特征存储为数据血缘（Data Lineage）提供了基础。数据血缘回答“这个模型是用什么 数据训练的？这个特征是从哪里派生的？”，在合规和调试场景中至关重要。 Amundsen（Lyft 开源）和 DataHub（LinkedIn 开源）是两个主要的元数据和血缘管理平台： •Amundsen 侧重数据发现（“我可以在哪里找到训练数据？”） •DataHub 侧重全面元数据（数据集、ML 模型、特征、仪表盘，通过统一的元数据图谱互连） 在 RAG 系统中，数据血缘追踪延伸至文档块级别：哪个文档块被检索用于生成某次回答？当文档更新后，哪些已经生 成的向量嵌入需要重新计算？这些都需要系统化的元数据管理。 •可复现性层次：实现 AI 工作流的可复现性，需建立以下层次：
4. 代码：Git（常规版本控制）
5. 环境：Docker 镜像 + conda/pip freeze（依赖锁定）
6. 数据：DVC/Delta Lake/LakeFS（数据版本管理）
7. 特征：Feast/Tecton（特征版本管理）
8. 模型：MLflow Model Registry（模型版本管理）
9. 参数：MLflow/W&B/Hydra（超参数记录） 六层的锁定共同构成一次实验的完整“快照”，可在任意时刻完整重现，这是 MLOps 的基础设施根基。

13.4 训练存储方案实战

以下设计一个完整的 Llama-70B 千卡训练存储方案，涵盖数据存储、预处理、训练检查点、推理模型仓库四大场景，包 含容量规划、性能基准和配置示例。

13.4.1 需求分析

训练配置： •模型：Llama-70B（80 layers, dim=8192, 64 heads） •并行策略：FSDP + TP8 + PP4 •GPU 规模：1024× A100-80GB（128 nodes × 8 GPUs） •数据集：3 TB tokenized 语料（不含原始数据） •Global batch size：约 4M tokens •检查点间隔：每 1000 步 •训练时长：预计 30 天（约 150K 步） 存储需求估算如表13-7所示。 表13-7 训练存储需求估算 存储项 单次大小 频率 月总量 需求 训练数据集 3 TB 静态 3 TB 高读带宽 检查点（全量） 约 2 TB 每 1000 步 约 300 TB (150 个全量) 高写带宽 (20+ GB/s) 检查点（保留） 每保留 3 个 最近 3 个 约 6 TB 快速恢复 日志与指标 约 1 GB/day 连续 约 30 GB 低带宽 Tokenizer 等工具 约 100 MB 静态 约 100 MB 低延迟 关键约束：

1. 检查点写入必须在 60 秒内完成（否则阻塞训练）
2. 数据加载不能成为训练瓶颈（GPU 计算的同时需完成下一 batch 的 I/O）
3. 集群网络分段：存储网络与计算网络物理隔离

13.4.2 存储架构设计

训练存储总体架构如图13-4所示，计算节点通过 RDMA 网络同时挂载并行文件系统与对象存储，节点本地 NVMe 承担数 据集缓存与检查点缓冲。 128-Node GPU Cluster Node 1: 8xA100 Node 2: 8xA100 Node 128: 8xA100 RDMA/RoCE 200Gbps Parallel File System - Lustre MDS: 2x NVMe OSS-1 OSS-2 OSS-3 OSS-4 NVMe Buffer: 4x3.84TB RDMA/RoCE NVMe Buffer: 4x3.84TB NVMe Buffer: 4x3.84TB Archive Object Store - MinIO (MDT: 2x3.84TB NVMe (OST1: 24x15.36TB NVMe) (OST2: 24x15.36TB NVMe) (OST3: 24x15.36TB NVMe) (OST4: 24x15.36TB NVMe) (MinIO-1: 12x15.36TB NVM (MinIO-2: 12x15.36TB NVM (MinIO-3: 12x15.36TB NVM (MinIO-4: 12x15.36TB NVM RAID1) e) e) e) e) 图13-4 Llama-70B 训练存储架构 Lustre 集群容量规划： •4 台 OSS，每台 24×15.36TB NVMe U.2 驱动器 •总 RAW 容量：4 × 24 × 15.36 TB ≈ 1,475 TB •使用 EC 8+4（效率 8/12 = 66.7%）：可用容量约 983 TB •单个驱动器顺序读写：3.5 GB/s 读，3.5 GB/s 写 •每 OSS 聚合带宽：24 × 3.5 GB/s ≈ 84 GB/s（受 200 Gbps × 4 NICs ≈ 100 GB/s 网络限制） •4 OSS 聚合带宽：约 400 GB/s（理论），约 250 GB/s（实测） MinIO 集群容量规划： •4 台节点，每台 12×15.36TB NVMe •总 RAW 容量：4 × 12 × 15.36 TB ≈ 737 TB •使用 EC:4（67% 效率）：可用容量约 492 TB •主要用于：数据集归档、检查点长期备份、模型仓库 节点本地 NVMe： •每计算节点 4×3.84 TB NVMe（RAID0），总容量约 15 TB/节点 •用途：本地数据集缓存、检查点 L2 缓冲、容器镜像存储 •128 节点总 NVMe 容量：约 1,920 TB

13.4.3 Lustre 配置

 # Requires Lustre 2.15+ server and client
 # MDS Configuration
 mkfs.lustre --mdt --fsname=ai-train --mgsnode=10.0.0.101@o2ib \
   --index=0 --backfstype=ldiskfs /dev/md0
 mount -t lustre /dev/md0 /mnt/mdt
 # OSS Configuration Example (48 OSTs total)
 for i in $(seq 1 12); do
      mkfs.lustre --ost --fsname=ai-train --mgsnode=10.0.0.101@o2ib \
        --index=$i /dev/nvme${i}n1
      mount -t lustre /dev/nvme${i}n1 /mnt/ost${i}

done

 # Client mount (GPU node)
 mount -t lustre 10.0.0.101@o2ib:/ai-train /lustre
 # Set training data directory striping

lfs setstripe -c 24 -S 4M /lustre/training-data/tokenized/

Set checkpoint directory striping

lfs setstripe -c 12 -S 2M /lustre/checkpoints/

Enable client caching

lctl set_param osc..max_dirty_mb=2048 # Client dirty cache lctl set_param osc..max_rpcs_in_flight=256 # Concurrent RPC

13.4.4 存储网络拓扑设计

AI 集群的存储网络遵循三网分离原则：计算网络（GPU-to-GPU）、存储网络（GPU-to-Storage）和管理网络使用物理独 立的网络平面。存储网络的拥塞控制是决定实际性能的关键因素。 在 RoCEv2 网络上，多对一写入模式（数百 GPU 同时向 Lustre OSS 写检查点）会触发 PFC（Priority Flow Control）风 暴，上游交换机因缓冲区溢出发送 PAUSE 帧，拥塞树向上游传播，影响不相关流量。缓解方案：

1. 配置 DCQCN（Data Center Quantized Congestion Notification），通过 ECN 标记在丢包前主动降速
2. 设置适当 PFC 阈值和 headroom buffer（通常为链路带宽 × 传播延迟 × 2）
3. Lustre LNet Multi-Rail 配置中将不同 OSS 流量分配到不同网络接口，避免单链路热点 在 InfiniBand 网络上，基于信用的链路层流控消除了 PFC 风暴风险，但需关注 SL（Service Level）到 VL（Virtual Lane）的映射，将存储流量分配独立 VL 避免与 NCCL 通信干扰。建议存储网络使用独立 Spine-Leaf 拓扑，Leaf 交换机 连接 GPU 节点，Spine 交换机连接存储 OSS，每个 Leaf 到 Spine 的上行带宽按 1:1 或 2:1 oversubscription 设计。

13.4.5 性能基准测试

在存储方案上线前，需进行严格的基准测试以确保满足需求：

 # Requires fio, OpenMPI (mpirun) and mdtest
 # 1. Single-node sequential read
 fio --name=seq-write --directory=/lustre/test \
   --rw=write --bs=4M --numjobs=8 --size=10G --group_reporting \
   --iodepth=64 --direct=1
 # 2. Multi-node concurrent write
 # Execute simultaneously on 128 nodes
 mpirun -np 128 -hostfile hosts.txt \
   fio --name=chkpt-sim --rw=write --bs=256M \
   --numjobs=1 --size=16G --group_reporting --direct=1 \
   --directory=/lustre/checkpoints/\$OMPI_COMM_WORLD_RANK/
 # 3. Metadata performance test
 mpirun -np 128 mdtest -d /lustre/mdtest -n 10000 -i 1
 # 4. Training data load simulation
 # Simulate 1024 ranks reading tokenized shards
 mpirun -np 1024 python -c "
 import torch, time
 data = torch.load(f'/lustre/training-data/shard_{rank}.pt')
 t = time.time()
 while True:
     data = torch.load(f'/lustre/training-data/shard_{rank}.pt')
     dt = time.time() - t
     print(f'Rank {rank}: {data.shape}, load_time={dt:.3f}s')
     t = time.time()

" 期望基准结果如表13-8所示。 表13-8 存储基准测试目标 测试项 目标 最小可接受 单节点顺序写 20 GB/s 10 GB/s 128 节点并发写 200 GB/s 聚合 100 GB/s 检查点窗口 (2 TB) < 20s < 60s 元数据创建 10K IOPS per node 5K IOPS per node 数据加载延迟 < 50 ms per batch < 200 ms per batch

13.4.6 训练框架集成

 # Requires DeepSpeed 0.14+, PyTorch 2.x
 # DeepSpeed configuration - checkpoint strategy
 ds_config = {

"train_batch_size": 4096, "checkpoint": { "tag_validation": "Warn", "load_universal": False, "use_node_local_storage": True, # Enable node-local NVMe "node_local_storage_path": "/nvme/checkpoints", "max_io_workers": 8, }, "data_sampling": { "enabled": True, "num_workers": 4, "curriculum_learning": {"enabled": False}, }, "tensorboard": { "enabled": True, "output_path": "/lustre/logs/tensorboard", "job_name": "llama-70b-train"

     }
 }
 # Training script - use node-local NVMe buffer
 from deepspeed.runtime.checkpoint_engine.torch_checkpoint_engine import TorchCheckpointEngine
 engine = deepspeed.initialize(model=model, config_params=ds_config, ...)
 # Save checkpoint: GPU -> node-local NVMe -> async sync to Lustre
 engine.save_checkpoint(
     save_dir="/lustre/checkpoints/step-%d" % global_step,
     tag="step-%d" % global_step,
     client_state={"step": global_step, "lr": lr},

)

13.4.7 监控与告警

Prometheus storage monitoring rules

groups:

• name: ai_storage_alerts rules:
  • alert: CheckpointWriteSlow expr: rate(lustre_ost_write_bytes_total[5m]) < 80e9 # < 80 GB/s for: 10m labels: severity: critical annotations: summary: "Checkpoint write bandwidth below threshold"
  • alert: LustreOSTNearFull expr: (lustre_ost_used_bytes / lustre_ost_total_bytes) > 0.85 for: 5m labels: severity: warning annotations: summary: "OST usage over 85%"
  • alert: NodeLocalNVMENearFull expr: (node_filesystem_avail_bytes{mountpoint="/nvme"} / node_filesystem_size_bytes{mountpoint="/nvme"}) < 0.10 for: 5m labels: severity: warning annotations: summary: "Node-local NVMe free space below 10%"

13.4.8 成本估算

年度存储总成本（不包括 GPU 计算节点自身的 NVMe）如表13-9所示。 表13-9 年度存储成本估算 组件 硬件成本 (USD) 年运维 (USD) 小计 (USD) Lustre OSS (4×24×NVMe) $400,000 $30,000 $430,000 Lustre MDS (2×NVMe) $15,000 $2,000 $17,000 200G InfiniBand 交换机 (4×) $160,000 $8,000 $168,000 MinIO 集群 (4×12×NVMe) $180,000 $15,000 $195,000 机柜/电力/冷却 — $40,000 $40,000 总计 $755,000 $95,000 $850,000 如果使用云服务（AWS FSx for Lustre + S3），等效年费约 $500,000-$800,000，但免去了硬件采购和维护。对于一次性 的 30 天 Llama-70B 训练任务，最经济的方案可能是 MinIO + 计算节点本地 NVMe + S3 归档，总成本约 $2-3 万（使用竞 价实例）。

13.4.9 关键决策清单

在方案确定前，逐项确认以下决策点： • 并行文件系统选型：Lustre (开源-HPC) vs WekaFS (商业-云原生) vs 3FS (定制-RDMA) • 检查点架构：纯并行FS 写入 vs 节点本地 NVMe 缓冲 vs S3 直写 • 网络拓扑：存储网络独立物理交换机，RoCEv2 或 InfiniBand • 分层策略：热数据（训练数据 + 最近 3 个检查点）在 Lustre，温/冷数据在 MinIO/S3 • 容量余量：至少预留 30% 的容量和带宽余量，应对数据增长和突发写入 • 灾难恢复：关键检查点在异地或另一个 AZ 有备份 • 监控覆盖：所有的 OST/MDT/OSS/MinIO 节点都有独立的 Prometheus exporter

13.5 数据预处理与加载

数据预处理管线将原始数据（网页文本、图像、音频等）转化为可直接供给模型训练的格式化数据，数据加载器则负责把 训练数据从磁盘或网络高效送入 GPU。本节覆盖端到端预处理管线架构、数据分片与全局 Shuffle 机制，以及主流数据加 载方案的对比。

13.5.1 数据预处理管线架构

AI 数据预处理管线（Data Preprocessing Pipeline）是将原始数据转化为可直接供给模型训练的格式化数据的工程系 统。对于 Llama 级别的训练（15T+ tokens），预处理管线的规模从“单机脚本”升级为“分布式数据处理系统”，需要处 理 TB-PB 级数据，应对数千个并发任务，保证最终输出的确定性和一致性。 •端到端管线全景：大规模预训练数据的处理流程通常包含以下阶段，如图13-5所示。 Raw Data Source Text Extraction Language Detection Quality Filter Deduplication Data Mixing Tokenization Sequence Packing Sharding Training Ready Common Crawl/Books/Wi HTML▶Text cld3/fastText Heuristic+Model Doc-level+Para-level Source Ratio BPE/SentencePiece Pack to fixed length Into N data shards Global index + Random or kipedia der 图13-5 端到端数据预处理管线 每个阶段有不同的计算特征：B-E 阶段的瓶颈主要在 CPU 和内存（大量字符串操作和模型推理），G-H 阶段的瓶颈在 I/O 和内存（字节级操作和序列重组），I 阶段纯粹是 I/O 和元数据整理。 •批量与流式管线：批量管线（Batch Pipeline）是 Apache Spark、Apache Beam 的经典模式。数据按固定的分片 （partition）并行处理，每个分片独立运行整个处理流程，最后合并结果。

 # Requires PySpark 3.4+
 from pyspark.sql import SparkSession
 spark = SparkSession.builder \

.appName("LLM-Data-Preprocessing")
.config("spark.executor.memory", "32g")
.config("spark.executor.cores", "8")
.getOrCreate()

 # read raw data
 raw_df = spark.read.text("s3://common-crawl/CC-MAIN-2024-*/wet.paths.gz")
 # quality filter
 filtered_df = raw_df \

.filter("length(text) > 100 AND length(text) < 100000")
.filter("language(text) == 'en'")
.filter("perplexity(text) < 500")
.filter("toxic_score(text) < 0.1")

 # deduplication (MinHash LSH)
 from spark_minhash_lsh import MinHashLSH
 dedup_df = MinHashLSH(filtered_df, input_col="text", threshold=0.8).deduplicate()
 # write to Parquet
 dedup_df.write.mode("overwrite").parquet("s3://dataset/processed/cc-filtered/")

批量管线的优势是吞吐量高，适合“先处理完再使用”的场景。劣势是延迟大，一个 PB 级数据集的全量处理可能需要数 小时到数天，不适合需要实时数据更新的场景。 流式管线（Streaming Pipeline）采用 Apache Kafka + Apache Flink/Beam Streaming 模式，数据持续流入、管线持续 处理，适合在线学习和增量数据更新场景。LLM 预训练通常是批量模式，但 RLHF 数据采集是天然的流式模式。 •Ray Data：Ray Data 是专为 ML/LLM 场景设计的分布式数据处理框架，相比 Spark 的优势在于与 ML 生态的原生集成 （与 Ray Train、Ray Serve 无缝互操作）。

 # Requires Ray 2.x
 import ray
 from ray.data import read_parquet
 # pipeline definition
 ds = ray.data.read_parquet("s3://dataset/raw/cc-filtered/")
 ds = ds \

.filter(lambda row: len(row["text"]) > 100)
.flat_map(tokenize_batch, fn_constructor_kwargs={"tokenizer_path": "tokenizer.model"}, num_cpus=0.5)
.map_batches(pack_sequences, batch_size=1000, num_cpus=1)
.repartition(4096) # repartition to training-usable count

write results

ds.write_parquet("s3://dataset/tokenized/llama-70b-train/") Ray Data 的独特优势：

1. 流式执行：算子间以流式方式传输数据块，不必等待所有上游数据完成
2. 弹性扩展：Ray 集群可动态增减 worker，管线自动重新平衡
3. 异构资源：支持为不同算子分配不同的 CPU:GPU 资源比 •主流开源语料与工具链：截至 2025 年，工业级预训练数据管线已高度工具化。值得对标的开源数据集包括： RefinedWeb（TII Falcon，约 5T tokens，证明纯网页经严苛过滤即可训练强模型）、Dolma（AI2，约 3T tokens，附 带完整处理工具 dolma toolkit）、RedPajama-Data-v2（约 30T tokens 附带质量信号）、FineWeb / FineWeb-Edu （HuggingFace，15T / 1.3T tokens）、DCLM（约 240T tokens 候选池 + 标准化评测）。 专用工具方面：HuggingFace datatrove 提供可组合的分布式处理块（提取、过滤、MinHash 去重、tokenize），原生 支持 Slurm 与本地/S3；NVIDIA NeMo Curator 将去重、质量分类、PII 脱敏等算子搬到 GPU，基于 RAPIDS cuDF/cuGraph 实现 GPU 版精确去重与模糊（MinHash+LSH+连通分量）去重，官方数据显示在多 GPU 上对数十 TB 语料的模糊去重较 CPU Spark 方案可提速一个数量级；Dask、Ray Data 则常作为底层调度。选型建议：CPU 密集的 启发式过滤与提取用 Spark/datatrove，去重与分类器打分等可 GPU 化的环节优先用 NeMo Curator 以压缩端到端时 长。 NeMo Curator GPU 去重性能量化：在 DGX H100 节点上对 2 TB Common Crawl 文本做 MinHash+LSH 模糊去重 （128 排列，Jaccard 0.8），端到端耗时约 45 分钟，吞吐约 6 GB/min/GPU。等价 CPU Spark 集群（100×16 核 executor）需 8-12 小时，GPU 方案提速 10-15 倍。加速来自 cuDF n-gram 哈希操作（GPU 并行整数运算）和 RAPIDS cuGraph 连通分量计算（避免 CPU 跨节点 shuffle）。若团队已有 GPU 集群，将去重迁移到 GPU 可将管线从 约 60 小时压缩到约 30 小时（混合：CPU 过滤 + GPU 去重）。 •管线 DAG 设计与调度：对于包含复杂依赖和分支的管线，DAG（有向无环图）定义和调度是关键。DVC 适用于文件级 DAG，而对于跨多阶段、需要分布式执行的复杂管线，Apache Airflow/Prefect/Dagster 是常见选择。

Requires Prefect 2.x

from prefect import flow, task @task def extract_common_crawl(date: str) -> str: ... @task def quality_filter(input_path: str) -> str: ... @task def deduplicate(input_path: str) -> str: ... @task def tokenize(input_path: str, tokenizer: str) -> str: ... @flow(log_prints=True)

 def llm_data_pipeline(date: str = "2024-01"):
     raw = extract_common_crawl(date)
     filtered = quality_filter(raw)
     unique = deduplicate(filtered)
     tokenized = tokenize(unique, "llama-tokenizer")
     return tokenized

•容错与 Exactly-Once 语义：大规模数据处理管线运行时间长（数小时到数天），节点故障不可避免。容错机制的核心 是 Checkpoint + 幂等输出：

1. 算子级检查点：Ray Data 和 Spark 自动保存中间结果的检查点，故障时从最近的检查点恢复，而非重新开始
2. 幂等输出：向最终数据分片写入时，使用确定性的文件名（如 shard-00001-of-02048.parquet ），支持覆盖写入
3. Exactly-Once 语义：严格的一次性输出在 AI 数据管线中并非必须，训练数据的一小部分重复对模型质量的影响可以 忽略。At-Least-Once + 去重后缀处理是更实用的选择 •性能优化要点： •数据格式：Parquet（列式，支持 Predicate Pushdown 和列裁剪）是预处理管线的最佳中间格式。 Protobuf/TFRecord 适用于特定框架（TensorFlow），但生态较窄 •I/O 并行度：分片数（partition count）应为 CPU 核心总数的 4-8 倍，确保负载均衡。过少的分片导致部分 worker 空闲，过多则元数据开销过大 •内存管理：在 tokenization 和序列 Packing 阶段，内存中需维护大量字符串。使用 Arrow（Apache Arrow）作为序 列化格式可减少 Python 对象的内存开销 •压缩：中间数据和最终训练数据都应压缩存储。ZSTD Level 3 提供良好的压缩比（2-3×）和吞吐（500+ MB/s per core） •多模态管线扩展：对于多模态数据（文本+图像+视频+音频），管线复杂性增加。每种模态通常需要独立的预处理分 支，在某个 join point 统一。预处理管线的设计和实现质量直接影响下游模型训练的效果和数据利用率。一个设计精良 的 PB 级管线可在 5-10 小时内完成全量预处理，一个未优化的管线可能耗时数天且频繁失败。

13.5.2 数据分片与全局 Shuffle

在大规模分布式训练中，数据分片（Sharding）和全局 Shuffle（Global Shuffle）直接影响训练收敛速度、模型质量和 GPU 利用率。不恰当的数据分布策略可能导致训练速度波动、某些 GPU 长期“饥饿”或模型收敛不充分。

1. 数据分片策略 Per-GPU Sharding 将数据集划分为 N 个不相交的分片，每 GPU 处理一个分片，这是数据并行训练（DDP）的默认数 据分布方式： gpu Dataset: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12] GPU 0: [1, 2, 3] GPU 1: [4, 5, 6] GPU 2: [7, 8, 9] GPU 3: [10, 11, 12] 每 GPU 处理固定子集，优点是实现极简，无需全局通信协调数据分配。缺点是如果训练步数 N ≫ 分片大小，模型可 能在该 epoch 内重复见到相同数据，造成过拟合风险。 steps Per-Node Sharding 按节点而非 GPU 分片。节点内的所有 GPU 共享同一分片，通过节点内高速 NVLink/NVSwitch 进行 数据分发，减少跨机数据传输。 全局分片（Global Sharding 或 Elastic Sharding）将数据集划分为大量小块（如 10,000 个 shards），每个训练步 （step）动态决定每个 GPU 读取哪些分片，提供最大的灵活性和负载均衡潜力，但需要全局协调机制。 对于 LLM 训练，通常采用 Per-Node Sharding + 节点内循环分片：数据分片数与节点数相等，节点内 GPU 通过循环 （round-robin）分配 micro-batch。
2. 确定性全局 Shuffle 在分布式训练中实现全局 shuffle 的困难在于：所有 rank 需要就 shuffle 后的全局数据顺序达成一致，但每个 rank 实际 只处理一部分数据。解决方法为基于种子的确定性 Shuffle。

 # deterministic global shuffle implementation
 import numpy as np
 class DeterministicGlobalShuffler:
     def __init__(self, num_samples: int, num_ranks: int, seed: int, epoch: int):
         self.num_samples = num_samples
         self.num_ranks = num_ranks
         self.seed = seed
         self.epoch = epoch
         # each rank uses same seed to generate same global permutation
         rng = np.random.RandomState(seed + epoch)
         self.global_permutation = rng.permutation(num_samples)
     def get_local_indices(self, rank: int) -> np.ndarray:
         # each rank extracts its own portion from global permutation
         # use contiguous sharding (rank 0: [0, N/R), rank 1: [N/R, 2N/R), ...)
         start = rank * self.num_samples // self.num_ranks
         end = (rank + 1) * self.num_samples // self.num_ranks
         return self.global_permutation[start:end]

关键机制：所有 rank 使用相同的种子和 epoch 号生成全局排列，然后各自取连续的索引段。这保证了三点，全数据集在 全局排列中恰好覆盖一次、每 rank 处理不相交的子集、不同 epoch 的全局排列不同（通过改变 epoch 号）。MosaicML StreamingDataset 和 Megatron-LM 的 --shuffle 标志均基于此机制。 3) 两阶段 Shuffle 对于超大规模训练（100B+ tokens），全量数据的全局 shuffle 可能导致两个问题：

1. 数据分片的物理存储与 shuffle 后的读取模式不匹配（随机读取跨分片的数据导致 I/O 性能急剧下降）
2. 全局排列的生成需要 O(N ) 内存存储索引表（对于 15T tokens 约 1.5TB 的索引空间） 两阶段 Shuffle 是应对方案： •阶段 1（Shard 级 Shuffle，粗粒度）：将数据集分成 M 个 shard（如 M = 10, 000），shuffle shard 的顺序 [S_3, S_1, S_7, S_2, ...] ，结果通过种子确定性生成 •阶段 2（分片内 Shuffle，细粒度）：每个 rank 加载分配到的 shard 批次，在本地对 shard 内的样本进行 shuffle，循环 读取时逐步将 buffer 中的样本喂给模型

 # Two-phase shuffle
 class TwoPhaseShuffler:
     def __init__(self, num_shards, shard_size, num_ranks):
         self.shards = list(range(num_shards))
         self.shard_size = shard_size
         self.shuffle_buffer = []
     def get_batch(self, rank, step, epoch):
         # Phase 1: determine which shard to read currently
         shard_order = self._get_shard_order(epoch)
         shard_idx = shard_order[(step // self.shard_size) % len(shard_order)]
        # Phase 2: extract samples from shard buffer on demand and shuffle locally
        if not self.shuffle_buffer:
            self._load_and_shuffle_shard(shard_idx)
        return self.shuffle_buffer.pop()

两阶段 shuffle 的近似质量：理论上不能生成完全随机的全局排列，但实验表明，当每 shard 样本数 N samples /M 足够小 （如 < 10,000）时，近似质量对模型收敛的影响可以忽略。 4) Streaming Shuffle 在流式场景下（新数据持续到达），Streaming Shuffle 使用滑动窗口实现部分 shuffle：

 # Streaming shuffle with sliding window
 class StreamingShuffler:
     def __init__(self, window_size=10000):
         self.window_size = window_size
         self.buffer = []
     def add(self, sample):
         self.buffer.append(sample)
         if len(self.buffer) >= self.window_size:
             pos = random.randint(0, len(self.buffer) - 1)

yield self.buffer.pop(pos) def flush(self): random.shuffle(self.buffer) yield from self.buffer Streaming shuffle 保证任意样本与其相邻样本的距离期望值为 window_size/2 ，打破了数据中的自然局部性（如连续 文本段落的主题相关性），但无法实现完全的全局随机性。对于 LLM 预训练，一个 window_size=100,000 的 streaming shuffle 可提供足够的 shuffle 质量。 5) 弹性数据集 弹性训练（Elastic Training，训练过程中动态增减 GPU 数量）需要数据集能动态调整分片分配。弹性数据集的关键设 计：

1. 细粒度分片：数据集被切分为远多于 GPU 数量的细粒度 shard（如 10,000 个 shard × 128 GPU）
2. 动态分片分配：当 GPU 数量变化时，重新计算 shuffle 排列并重新分配 shard
3. 分片级检查点：记录每个 shard 的消费进度，支持 GPU 数量变化后的状态恢复 TorchElastic（PyTorch）和 DeepSpeed 的 Elastic Training 均内置了弹性数据集的机制。

6. 分片大小优化 分片大小（Shard Size）的优化是数据加载性能的关键调优点： •过大分片（如 1 GB/shard）：sequential read 高效，但加载一个 shard 需要长时间（10+ 秒），shuffle 粒度粗， shuffle buffer 覆盖不充分 •过小分片（如 1 MB/shard）：shuffle 粒度细，但持续的 open()/close() 产生高元数据开销，I/O 吞吐下降 经验法则：分片大小应为 batch_size × 100-1000 样本量级。对于 LLM text 数据，每个 shard 约 100 MB-500 MB 是较好 平衡点。对于图像数据，每个 shard 约 1-10 GB（对应数万张图片）是常见范围。

13.5.3 高性能数据加载器对比

数据加载器（DataLoader）是训练数据从磁盘/网络进入 GPU 计算的桥梁，其效率直接影响 GPU 利用率和训练吞吐。一 个性能不佳的 DataLoader 会导致 GPU 空闲等待（Data Stall），即使使用最强大的 GPU 集群，MFU 也会被拉低 10-30 个百分点。以下对比五大主流数据加载方案。

1. PyTorch DataLoader PyTorch 原生 DataLoader 是最广泛使用的方案，基于多进程预取（multi-processing prefetching）：

 # Requires Python 3.10+ and PyTorch 2.x
 from torch.utils.data import DataLoader
 loader = DataLoader(

dataset,

     batch_size=64,
     num_workers=8,           # number of worker processes
     pin_memory=True,          # copy data to page-locked memory (accelerates CPU-to-GPU transfer)
     prefetch_factor=2,        # number of batches prefetched per worker
     persistent_workers=True, # keep workers alive across epochs (reduce fork overhead)
     drop_last=False,

) 核心参数解析： • num_workers ：每个 worker 是一个独立进程，通过 fork() 创建。worker 负责从磁盘读取数据并做 CPU 上的预处 理（解码、裁剪、归一化）。worker 数量通常设置为 4-16，取决于 CPU 核心数和 I/O 瓶颈 • pin_memory=True ：将 batch 数据放入页锁定（page-locked）内存区域。页锁定内存可直接被 GPU 的 DMA 引擎访 问， host-to-device 传输速度提升约 2-3×。代价是页锁定内存有限（通常几十 GB） • prefetch_factor ：每个 worker 预加载的 batch 数。 prefetch_factor=2, num_workers=8 意味着最多有 16 个 batch 在内存中等待 GPU 使用。增加此值可改善 GPU 利用率，但占用更多 CPU 内存 PyTorch DataLoader 的局限：

1. Python GIL 干扰：虽然 worker 是独立进程，但 batch 组装（collate_fn）在主进程中执行，受 GIL（Global Interpreter Lock）约束
2. 序列化开销：worker 进程 → 主进程的数据传输通过 multiprocessing.Queue ，涉及 pickle 序列化和跨进程 IPC， 对于大 batch size 时不容忽视
3. 无远程 I/O：原生 DataLoader 仅支持本地文件系统。对 S3/HTTP 等远程数据源需自行实现

2. WebDataset WebDataset 将大量小文件（图像、文本片段）打包成 .tar 归档文件，以分片（shard）级别进行随机访问。核心思 想：文件系统对小文件不友好（ open() 和 stat() 开销），而 tar 文件内的顺序读取极快。

 # Requires webdataset 0.2.x
 import webdataset as wds
 # build WebDataset pipeline
 dataset = (
     wds.WebDataset("s3://dataset/imagenet-train-{0000..1280}.tar")

.shuffle(1000) # shuffle at shard level .decode("pil") # decode images .to_tuple("jpg", "cls") # extract as (image, label) tuple .map(transform) # apply data augmentation ) loader = wds.WebLoader(dataset, batch_size=64, num_workers=8) WebDataset 的优势： •Shard 级随机访问：通过随机选择 tar 文件实现近似全局 shuffle，无需加载全量元数据 •远程 I/O 原生： s3:// 、 http:// 和 gs:// 协议原生支持，适合云存储训练 •I/O 友好：顺序读取 tar 文件中数百个文件连续地址，远快于独立打开数百个文件 WebDataset 的劣势： shuffle(1000) 将 1000 个连续的样本放入 shuffle buffer，并非真正的全局 shuffle。对于 LLM 训练，这种近似 shuffle 通常足够好。 3) MosaicML StreamingDataset StreamingDataset 是 MosaicML（现 Databricks Mosaic）的核心贡献，提供确定性全局 Shuffle 与远程流式传输：

 # Requires streaming (MosaicML) 0.7.x
 from streaming import StreamingDataset
 dataset = StreamingDataset(
     local="/tmp/cache/llama-data",   # local cache directory
     remote="s3://my-bucket/llama-data", # remote dataset
     shuffle=True,                     # deterministic shuffle
     shuffle_algo="py1b",             # shuffle algorithm
     batch_size=64,

) 数据集被预先分片为 MDS（Mosaic Data Shard）格式，每个分片包含多个样本。确定性 Shuffle 的核心价值在于：多机 多卡场景下，所有 rank 无需通信即可获得一致的全局数据分布，每个 rank 独立计算自己应处理的数据范围，无需中央调 度器或全局协调。 远程流式传输：StreamingDataset 在需要某分片时才从 S3 下载该分片到本地缓存。当本地缓存接近容量上限时，LRU 策略淘汰旧分片。这种按需加载模式使训练可在几 TB 的数据集上运作，而无需存储全量本地副本。 4) NVIDIA DALI DALI（Data Loading Library）将数据预处理卸载到 GPU，实现“数据在 GPU 上就绪，直接供训练使用”的零拷贝流 程：

Requires NVIDIA DALI and CUDA 12.x

from nvidia.dali import pipeline_def, fn @pipeline_def(batch_size=64, num_threads=4, device_id=0) def dali_pipeline(): jpegs, labels = fn.readers.file( file_root="/data/imagenet/train", random_shuffle=True, )

     images = fn.decoders.image(jpegs, device="mixed") # GPU decoding
     images = fn.resize(images, resize_x=224, resize_y=224)
     images = fn.crop_mirror_normalize(

images, mean=[0.485255, 0.456255, 0.406255], std=[0.229255, 0.224255, 0.225255], )

     return images, labels
 pipe = dali_pipeline()
 pipe.build()

DALI 的关键特性： •GPU 预处理：JPEG 解码、图像缩放、归一化均在 GPU 上并行执行，避免 CPU → GPU 的数据拷贝 •算子融合：多个图像变换算子可在 GPU 上自动融合为单个 kernel •Pipeline 并行：数据加载、解码、预处理的阶段可流水线执行，吞吐不受单阶段延迟限制 DALI 主要适用于计算机视觉任务，对 NLP/LLM 的支持有限（文本处理的最佳路径通常仍是 CPU 上的 tokenization）。 5) FFCV FFCV 通过预编译数据集和 JIT 数据预处理实现极限 I/O 性能：

 # Requires FFCV 0.0.3+ and CUDA
 from ffcv.loader import Loader, OrderOption
 from ffcv.fields import IntField, RGBImageField
 from ffcv.writer import DatasetWriter
 # 1. compile raw data to compact format
 writer = DatasetWriter("cifar_train.beton", {"image": RGBImageField(), "label": IntField()})
 writer.from_indexed_dataset(original_dataset)
 # 2. fast loading
 loader = Loader(

"cifar_train.beton",

     batch_size=64,
     num_workers=8,
     order=OrderOption.RANDOM,
     os_cache=True,          # leverage OS page cache
     pipelines={

"image": [compiled_transform], "label": [IntDecoder()], }, ) FFCV 的性能秘密：

1. 编译式预处理：用户定义的数据变换被 JIT 编译为高效的机器码（通过 NumPy C API），绕过 Python 解释器
2. OS 页缓存利用：通过 os_cache=True 引导内核将数据集文件保持在页缓存中。结合大内存服务器（512 GB+ RAM），全量数据集可被缓存到内存
3. 紧凑存储格式：.beton 格式经过字节级优化，I/O 带宽利用接近硬件上限 在 ImageNet 基准上，FFCV 在 8×A100 上达到 500K+ 图像/秒的吞吐，远超其他 DataLoader（通常在 50K-200K 图像/ 秒范围）。 五大方案的性能对比如表13-10所示。 表13-10 数据加载器综合对比 DataLoader 吞吐 (images/s, A100×8) GPU 利用率 远程 I/O 确定性 Shuffle NLP 支持 适用场景 PyTorch DL 80K-150K 60-85% 需自定义 手动 原生支持 通用基准线 WebDataset 150K-250K 75-92% S3/HTTP 原生 近似 支持 云存储 + CV StreamingDataset 120K-200K 70-88% S3/GCS 原生 确定性 最好 (MDS) 大规模 LLM NVIDIA DALI 250K-400K 88-95% 有限 需自定义 有限 CV 专用 FFCV 400K-550K 90-97% 无 手动 有限 极致 CV 性能 对于 LLM 预训练，MosaicML StreamingDataset 和 WebDataset 是最具实操价值的选择。对于中小型 CV 训练， PyTorch DataLoader + NVIDIA DALI 提供足够性能。对于极致性价比的 CV 训练，FFCV 是无与伦比的选择。

13.6 数据质量与配比

数据质量是决定大语言模型性能的最重要因素之一。Llama 3 技术报告强调，数据质量与去重对最终模型能力具有决定性 作用，高质量数据的价值往往超过单纯堆叠更大的数据量。本节覆盖质量过滤、去重技术与数据配比、课程学习两类主 题。

13.6.1 数据质量过滤

质量过滤和去重是数据管线中最关键也最计算密集的步骤，其目标是从海量原始语料中筛出高质量训练样本。 •语言检测：对于多语言预训练，需确定每个文档的主要语言。常用工具：

 # Requires pycld2
 import pycld2 as cld2   # Compact Language Detector 2
 text = "The quick brown fox jumps over the lazy dog."

is_reliable, text_bytes_found, details = cld2.detect(text)

keep only high-confidence

def filter_english(text, confidence_threshold=0.9): reliable, _, details = cld2.detect(text) # details[0] = (language_name: str, language_code: str, percent: int, score: float) return reliable and details[0][2] > 95 and details[0][3] > confidence_threshold pycld2 / pycld3 基于 Google 的 Compact Language Detector，速度极快（μs 级），准确率高于 fastText 和 langdetect。 •困惑度过滤：使用一个小型语言模型（如 KenLM 5-gram 模型）评估文本的自然度。困惑度过高（如 > 1000）的文本 通常是无意义字符序列、乱码或格式破损的文本。

 # Requires kenlm and a 5-gram arpa model
 import kenlm
 model = kenlm.Model("ccnet.en.arpa.bin")
 def perplexity_filter(text, max_ppl=500):
     score = model.score(text)
     num_tokens = len(text.split())
     if num_tokens == 0:
         return False
     ppl = 10 ** (-score / num_tokens)
     return ppl < max_ppl

对于 Common Crawl 数据，perplexity 阈值为 500-1000 可过滤掉约 30-50% 的低质量网页文本。 •毒性分类：使用 Jigsaw Perspective API 或 Detoxify 等预训练分类器识别和过滤含有有害内容的文本。

 # Requires detoxify and transformers
 from detoxify import Detoxify
 detox = Detoxify("original")
 def toxicity_filter(text, max_toxicity=0.1):
     results = detox.predict(text)
     return max(results.values()) < max_toxicity

•启发式规则：一组简单但有效的统计规则：

 def heuristic_filters(text):
     # length filter
     if len(text) < 100 or len(text) > 100000:
         return False
     # symbol ratio -- garbled text often has abnormally high symbol ratio
     alpha_ratio = sum(c.isalpha() for c in text) / max(len(text), 1)
     if alpha_ratio < 0.4:
         return False
     # line dedup -- machine-generated text may have many duplicate lines
     lines = text.split('\n')
     unique_ratio = len(set(lines)) / max(len(lines), 1)
     if unique_ratio < 0.3:
         return False
     # dirty word count -- measure quality gap from Wikipedia
     if text.count("lorem ipsum") > 0:
         return False
     return True

综合运用上述过滤器，从 Common Crawl 原始数据到高质量训练数据的通过率通常在 5-20% 之间。 •模型驱动的质量分类器：2024 年以来，头部开源与闭源数据管线普遍以“轻量分类器打分”取代或补强启发式规则， 因为规则难以捕捉“教育价值/知识密度”这类语义信号。三条代表性路线：

1. FineWeb-Edu（HuggingFace）：用 Llama-3-70B-Instruct 对约 46 万网页标注 0-5 的“教育价值”分数，训练一个 基于嵌入 + 线性回归头的分类器，对 FineWeb 打分后仅保留 score≥3 的样本，得到约 1.3T tokens 的 FineWeb- Edu，在 MMLU、ARC 等基准上显著优于同量级未过滤语料
2. DCLM（DataComp-LM）：用 fastText 二分类器区分“高质量指令/问答文本（OpenHermes、ELI5）”与随机 Common Crawl，仅保留打分靠前的样本
3. Llama 3：采用 fastText 与基于 DistilRoBERTa 的质量分类器级联，并用 Llama 2 生成的“是否高质量”判别信号作 为训练标签 工程上，分类器推理成本远高于启发式规则（需 GPU 批量前向），因此通常放在启发式过滤/去重之后作为最后一道 “精筛”，整体使 Common Crawl 通过率再降至个位数百分比。

13.6.2 去重技术

•精确匹配去重：对每个文档计算哈希值（MD5/SHA256），相同哈希的文档为重复。最简单、最快，但只能检测完全相 同的文档，对轻微修改（多个空格、一行差异）完全无效。

 import hashlib
 from collections import defaultdict
 def exact_dedup(texts):
     seen = set()
     unique = []
     for text in texts:
         h = hashlib.sha256(text.encode()).hexdigest()
         if h not in seen:
             seen.add(h)
             unique.append(text)
     return unique

•MinHash 与 LSH：MinHash + LSH（局部敏感哈希）用于检测近似重复（near-deduplication）。MinHash 将文档的 n-gram 集合压缩为固定大小的签名（signature），LSH 将签名分桶，同一桶内的高相似度文档候选对。

 # Requires datasketch
 from datasketch import MinHash, MinHashLSH
 # create LSH index
 lsh = MinHashLSH(threshold=0.8, num_perm=128)
 # create MinHash signature for each document
 for i, text in enumerate(docs):
     m = MinHash(num_perm=128)
     for word in text.split():
         m.update(word.encode('utf8'))
     lsh.insert(i, m)
 # query near duplicates
 near_duplicates = lsh.query(lsh.keys.get(doc_id))

MinHash 的 Jaccard 相似度近似精度由 num_perm （排列数）控制：128 个排列时标准误差约 0.05，256 个排列时约 0.035。对于 PB 级去重，128 个排列是计算效率和精度的最佳平衡。 •FuzzyMinhash 与分层去重：2024 年以来，工业级去重管线已收敛到三层分层架构：

1. 第一层：精确 URL/SHA256 去重（Bloom filter），去除 10-30% 的 Common Crawl 数据
2. 第二层：MinHash+LSH 近似去重。FuzzyMinhash（Briggs & Chen, 2024）利用低精度模糊哈希在同等内存下将排 列数从 128 扩展到 512-1024，Jaccard 估计标准误差降至 ±0.015-0.02，召回率从约 85% 提升到约 95%
3. 第三层：精确 Jaccard 或后缀数组子串去重（仅对 LSH 候选对，计算量从 N² 降至 N×k），假阳性率 <1% 在此补救 三层总开销比纯 LSH 仅增加 20-30%，去重质量提升 8-12%。RedPajama-Data-v2 和 Dolma 均采用类似层级架构。 •SimHash：SimHash（Google）将高维特征向量降为固定长度的二进制指纹。其独特性质是海明距离（Hamming distance）与两向量夹角成正比（即与余弦相似度成反比，相似度越高海明距离越小），使其适合大规模近似去重。 Google 使用 SimHash 来去重整个 Web 索引。 •后缀数组：后缀数组（Suffix Array）通过构建文档集合的后缀数组来检测精确的子串重复。算法复杂度 O(N log N )（ N 为总字符数）。能检测到任意长度的精确子串重复，但无法处理近似匹配，且内存占用约为原始文本的 4-8 倍（仅 SA，使用 32-bit 整数）或 8-13 倍（含完整辅助结构 LCP 与 ISA）。 •SemDeDup 语义去重：SemDeDup（2023, Stanford/Meta）将去重从字面/近似级别提升到语义级别。使用预训练嵌 入模型（如 CLIP/Sentence-BERT）将每个样本编码为向量，计算余弦相似度矩阵，删除与已有样本语义高度一致的样 本。

 # Requires sentence-transformers
 from sentence_transformers import SentenceTransformer, util
 model = SentenceTransformer("all-MiniLM-L6-v2")     # lightweight embedding model
 def semantic_dedup(texts, threshold=0.95):
     embeddings = model.encode(texts, convert_to_tensor=True)
     # compute cosine similarity between all pairs
     cosine_scores = util.cos_sim(embeddings, embeddings)
     to_keep = set(range(len(texts)))
     for i in range(len(texts)):
         if i not in to_keep:

continue

         for j in range(i+1, len(texts)):
             if cosine_scores[i][j] > threshold:
                 to_keep.discard(j) # keep i, remove j
     return [texts[i] for i in sorted(to_keep)]

语义去重的计算复杂度是 O(N ) 的 embedding 比较，对于百万级样本是可行的，但对于十亿级需要降级，通常先通过 MinHash 聚类（将十亿文档降为百万级候选对），再对每组候选对执行语义去重。 分布式去重工程 对于 TB/PB 级数据集的去重，单机方法不可行，需要分布式计算框架：

 # Requires PySpark 3.x
 from pyspark.ml.feature import MinHashLSH, NGram, HashingTF
 # 1. generate n-gram features
 ngram = NGram(n=5, inputCol="text", outputCol="ngrams")
 df = ngram.transform(raw_df)
 hashingTF = HashingTF(inputCol="ngrams", outputCol="features", numFeatures=10000)
 df = hashingTF.transform(df)
 # 2. MinHash LSH
 mh = MinHashLSH(inputCol="features", outputCol="hashes", numHashTables=5)
 # 3. approximate dedup
 approx_similar = mh.approxSimilarityJoin(df_a, df_b, threshold=0.8, distCol="jaccard")

性能指标参考（使用 1000× Spark executor, 每 executor 4 cores + 16 GB RAM）： •1 TB 文本数据的 MinHash 特征提取：约 2-3 小时 •LSH 分桶与相似对计算：约 1-2 小时 •总去重处理吞吐：约 100 GB/hour per 100 executors 去重对模型质量的影响 消融实验反复证明：从训练集中移除低质量数据和近似重复数据，对下游任务的提升往往等同于增加 10-30% 的模型参数 量。具体量化（Meta AI 内部研究）： •去重可将模型困惑度降低 3-5% •质量过滤可将 MMLU 得分提升 2-4 个百分点（同等训练 token 量） •语义去重（SemDeDup）在额外去重之外，对长尾事实性知识的覆盖几乎没有负面影响（因为语义去重仅在高度相似 的样本间进行） 质量过滤和去重是“一次投入、持续回报”的基础设施投资，处理过的数据集可复用于多次训练和多个模型，显著提升每 个训练 token 的“知识密度”。评测集去污染（防泄漏）是另一项独立的数据治理措施，防止训练语料混入评测原题导致 分数虚高。

13.6.3 数据配比与课程学习

在 LLM 预训练中，数据集通常由多个异质来源拼接而成，包括 Wikipedia（百科全书知识）、Books（长文本连贯性）、 Common Crawl（互联网多样性）、Code（逻辑推理）、arXiv/PubMed（专业知识）等。如何确定这些来源的混合比 例，以及训练过程中如何调整比例，是决定模型能力分布的关键因素。 •静态混合：静态混合（Static Mixing）在整个训练过程中使用固定的数据来源比例。Llama 1 的经典配比约为 Common Crawl 67.0% + C4 15.0% + GitHub 4.5% + Wikipedia 4.5% + Books 4.5% + ArXiv 2.5% + StackExchange 2.0%。每个 mini-batch 按此比例随机采样。 实现方式：为每个来源预先分配权重，在每个 batch 中按权重采样：

     import numpy as np
     sources = ["cc", "c4", "github", "wiki", "books", "arxiv", "stackexchange"]
     weights = [0.670, 0.150, 0.045, 0.045, 0.045, 0.025, 0.020]
     def sample_batch(batch_size, dataset_loaders):
         # select source for each sample (by weight)
         source_choices = np.random.choice(len(sources), size=batch_size, p=weights)
         batch = []
         for sid in source_choices:
             batch.append(next(dataset_loaders[sources[sid]]))
         return batch

静态混合的局限：无法适应训练过程中的能力演化。例如，代码能力在训练后期受益更多，但固定的比例无法反映这种时变性。 •动态混合：动态混合（Dynamic Mixing）在训练过程中调整来源比例。常见策略： •分段函数：前 70% 步用比例 A，后 30% 步用比例 B •退火（Annealing）：某些高质量来源（如 Wikipedia）的比例随训练步数线性降低，粗糙数据占比提升 •课程退火：先以高质量数据训练（建立基础能力），再逐步引入低质量但更多样化的数据（扩展知识面）

 def dynamic_weights(sources, step, total_steps):
     # first 70% steps: high proportion of high-quality data
     # last 30% steps: gradually anneal to final ratio
     if step < 0.7 * total_steps:
         return {"wiki": 0.15, "books": 0.10, "cc": 0.55, "code": 0.10, "others": 0.10}
     else:
         progress = (step - 0.7 * total_steps) / (0.3 * total_steps)
         wiki_w = 0.15 * (1 - 0.5 * progress) # wiki anneals from 15% to 7.5%
         books_w = 0.10 * (1 - 0.5 * progress) # books anneals from 10% to 5%
         cc_w = 1.0 - wiki_w - books_w - 0.15 - 0.10
         return {"wiki": wiki_w, "books": books_w, "cc": cc_w, "code": 0.15, "others": 0.10}

Llama 3 技术报告揭示了一个重要发现：在训练末期（最后数百万步）小幅退火高质量数据的比例，可以在不显著牺牲基 础能力的前提下释放更多“学习容量”给多样化的互联网数据。 •DoReMi 数据配比优化：DoReMi（2023, Stanford）通过训练一个微型代理模型来自动学习最优数据混合配比，避免 在完整模型上执行昂贵的消融实验。算法流程：

1. 训练一个小的“参考模型”（如 280M 参数）在均匀混合的所有数据来源上
2. 同时训练一个“代理模型”，其数据采样权重由 DRO（Distributionally Robust Optimization）优化
3. DRO 的目标：采样权重使代理模型在所有数据域上的超额损失（excess loss，相对于参考模型）最小化
4. 收敛后的采样权重被用作大型模型的静态配比 DoReMi 在 The Pile 与 GLaM 数据集上验证：用 280M 参考/代理模型优化域权重后，训练的 8B 主模型在下游任务上超 过在手调配比上训练的同等模型。对于 70B+ 模型，DoReMi 通过几百美元的代理模型训练（280M 模型训练约 10 GPU-小时），节省了数千美元的消融实验成本。 •DoGE 梯度数据选择：DoGE（2024, EPFL）从数据对模型梯度的影响角度来优化数据选择。核心思想：并非所有训练 样本的梯度更新对模型参数同等有用。某些样本的梯度与全量数据的平均梯度方向高度一致，这些样本具有更高“学习 价值”。 DoGE 在每个训练步中：
5. 从候选数据池中采样一个小批量
6. 计算每个样本梯度与参考梯度（从验证集计算）的对齐分数（alignment score）
7. 仅保留 top-K 最高对齐分数的样本进行实际参数更新 实践中，DoGE 的过滤比例通常为 20-50%，即丢弃一半候选样本，但模型收敛速度提升 1.5-2×。DoGE 对长尾数据的 负影响很小（因为长尾但高质量的样本往往有高对齐分数）。 •课程学习：课程学习（Curriculum Learning）从“简单到困难”的顺序安排训练数据，灵感来源于人类学习过程。在 LLM 预训练中，课程学习的维度包括： •文本复杂度：先训练短文本（256 tokens 序列），再逐步增长至完整序列（4096/8192 tokens）。这在训练初期显著降 低计算开销（计算复杂度与序列长度平方成正比），同时让模型先建立局部语言模式，再捕捉长程依赖 •来源难度：Wikipedia（规范化、高质量）→ Books（长文本、连贯性）→ Common Crawl（多样但噪声大）→ Code （结构化、逻辑性强）

 # curriculum learning scheduler
 class CurriculumScheduler:
     def __init__(self, total_steps):
         self.phase_boundaries = {

0: {"max_seq_len": 256, "sources": ["wiki", "books"]}, 0.2: {"max_seq_len": 512, "sources": ["wiki", "books", "cc_high_quality"]}, 0.5: {"max_seq_len": 1024, "sources": ["wiki", "books", "cc", "code"]}, 0.8: {"max_seq_len": 2048, "sources": ["cc", "code", "arxiv", "all"]}, 0.9: {"max_seq_len": 4096, "sources": ["all"]},

         }
     def get_curriculum(self, step, total_steps):
         progress = step / total_steps
         phase = max(k for k in self.phase_boundaries if progress >= k)
         return self.phase_boundaries[phase]

数据重复与 Epoch 数 传统深度学习倾向于控制 epoch 数（1-100 epochs），避免过拟合。但 LLM 预训练中，数据的“计算最优复用次数”是 一个仍在研究中的问题。Chinchilla 缩放定律表明：给定计算预算 C ，训练 tokens 数应与模型参数数量同比例增长，这 暗示每个 token 平均仅被“看到”约 1-4 次。 Data Mixing Laws 与退火 Ye 等人（2024, NeurIPS）的 Data Mixing Laws 首次量化了数据混合比例与模型性能的关系：两域混合时验证损失服从 L(p) = c + k ⋅ p + k ⋅ (1 − p) ，α 越大表示该域边际贡献递减越快（Wikipedia α ≈ 0.25，Common Crawl α ≈ −αA −αB 0.4 − 0.5）。退火（Annealing）方面，DeepSeek-V2 在最后 10% 训练步中将高质量数据占比从 15% 提升至 40%，同时 A B 序列长度从 4K 提升到 32K，最终在 MATH 和 HumanEval 上分别获得 8.1 分和 5.3 分的额外提升。Llama 3-405B 使用了 专门的退火数据集（约 4000 万 token, 约 0.03% 训练量），MMLU 提升约 1 个百分点，高质量数据的梯度信号能在训练末 期做精细的参数校准。 实践中，Llama 2 和 Llama 3 的数据重复率约 4-8 epochs（即约等于总训练数据被重复 4-8 次使用）。对于重复次数更高 的数据子集（如 Wikipedia 可能被重复 50+ 次），过拟合的风险被 LLM 的极大参数量所缓解，模型对单一特定数据集的记 忆需要大量重复才能形成。 一项 Meta 内部研究发现：数据重复增加 2× 对模型质量的影响等价于模型参数增加约 15%，但计算成本仅增加 2×。因 此在实际资源限制下，优先增加数据量（即使数据重复度较高）通常优于过度增大模型。 实用配比建议 基于多个开源模型的训练实践（Llama 3, Qwen, DeepSeek, Yi），针对通用能力的 LLM 预训练，推荐起始配比如表13-11 所示。 表13-11 LLM预训练起始配比建议 来源 建议比例 作用 Common Crawl (高质量过滤后) 55-65% 知识广度、语言多样性 Code (开源仓库) 10-15% 推理能力、格式化输出 Books 5-8% 长文本连贯性、叙事能力 Wikipedia 3-5% 事实准确性、百科全书知识 Academic Papers (arXiv/PubMed) 3-5% 专业知识、科学推理 Web Documents (高质量) 5-8% 多样化知识来源 Multilingual (非英语) 5-8% 多语言能力 Conversation/Instruction 2-4% 指令遵循能力 本章节数据配比参考：Chinchilla Scaling Laws (Hoffmann et al., 2022), DoReMi (Xie et al., 2023), DoGE (Fan et al., 2024), Llama 3 Technical Report (Meta, 2024)。

13.7 多模态数据与数据飞轮

多模态模型（Vision-Language Models, VLMs）如 GPT-4V、Gemini、LLaVA、Qwen-VL 等需要处理跨越文本、图像、 视频和音频的异构数据，其数据管线的设计目标是高吞吐加载、低 IO 阻塞、解码开销可控。数据飞轮则是模型从部署中 持续改进的自我增强机制。本节覆盖多模态数据预处理、视频数据管线、数据飞轮与在线学习，以及后训练数据的验收与 标注体系。

13.7.1 多模态数据预处理

1. 图文数据预处理 图文对（Image-Text Pairs）是多模态预训练的基础数据来源。典型来源包括 LAION-5B、DataComp、COYO-700M 等大 规模图文数据集。 图像处理管线：

 # Requires Python 3.10+, PyTorch and torchvision
 from PIL import Image
 from torchvision import transforms
 def preprocess_image(image_path, target_size=(336, 336)):
     """

Pipeline: decode -> validate -> resize -> normalize """ try: img = Image.open(image_path).convert("RGB") except (IOError, OSError): return None # corrupted image, discard directly w, h = img.size

     if w < 50 or h < 50:       # filter out undersized images
         return None
     if w / h > 5 or h / w > 5: # filter out extreme aspect ratios
         return None
     transform = transforms.Compose([
         transforms.Resize(target_size, interpolation=transforms.InterpolationMode.BICUBIC),
         transforms.CenterCrop(target_size),
         transforms.ToTensor(),
         transforms.Normalize(mean=[0.48145466, 0.4578275, 0.40821073],
                            std=[0.26862954, 0.26130258, 0.27577711]),

]) return transform(img) 文本处理管线：

 def preprocess_caption(caption, tokenizer, max_length=77):
     """
     CLIP-style text encoding: truncate or pad to fixed length
     """
     tokens = tokenizer(

caption,

         padding="max_length",
         truncation=True,
         max_length=max_length,
         return_tensors="pt",

) return tokens 图文对齐质量过滤：使用 CLIPScore（图文嵌入的余弦相似度）进行过滤：

 # Requires transformers
 from transformers import CLIPModel, CLIPProcessor
 model = CLIPModel.from_pretrained("openai/clip-vit-large-patch14")
 processor = CLIPProcessor.from_pretrained("openai/clip-vit-large-patch14")
 def clip_score_filter(image, caption, min_score=0.25):
     inputs = processor(text=[caption], images=image, return_tensors="pt", padding=True)
     outputs = model(**inputs)
     score = outputs.logits_per_image.item() # image-text matching score
     return score > min_score

LAION-5B 从约 500 亿候选图文对（Common Crawl 抓取）中用 CLIP 相似度阈值（英文子集 0.28、多语种 0.26）过 滤，最终保留约 58.5 亿对高质量图文对。 2) 音频预处理 音频模态的处理通常包括：

 # Requires torchaudio
 import torchaudio
 import torchaudio.functional as F
 def preprocess_audio(audio_path, target_sample_rate=16000, duration=10.0):

waveform, sr = torchaudio.load(audio_path)

     # resample to target sample rate
     if sr != target_sample_rate:
         waveform = F.resample(waveform, sr, target_sample_rate)
     # truncate or pad to fixed duration
     target_length = int(target_sample_rate * duration)
     if waveform.size(1) > target_length:
         waveform = waveform[:, :target_length]
     else:
         waveform = torch.nn.functional.pad(waveform, (0, target_length - waveform.size(1)))
     # compute log-Mel spectrogram (Whisper style)
     mel_spec = F.melspectrogram(

waveform,

         sample_rate=target_sample_rate,
         n_fft=400,
         hop_length=160,
         n_mels=80,

) log_mel = torch.log(mel_spec + 1e-10) return log_mel 对于音频-语言模型（如 Qwen-Audio），音频通常被转换为离散 token（通过 EnCodec 或类似编码器），与文本 token 共 同输入 Transformer。 3) 统一多模态 Tokenization 多模态模型的趋势是使用统一的 token 化方案，将所有模态表示为同质 token 序列，然后喂给一个统一的 Transformer。这简化了管线设计，因为在 token 化之后所有模态的预处理逻辑统一。

 # pseudocode: unified multimodal tokenization
 def multimodal_tokenize(image, text, audio=None):
     tokens = []
      # image tokens: [IMG_START] + [IMG_PATCH_1] + ... + [IMG_PATCH_N] + [IMG_END]
      if image is not None:
          tokens.append(IMG_START_TOKEN)
          image_patches = vision_encoder.encode_image_to_tokens(image)
          tokens.extend(image_patches)
          tokens.append(IMG_END_TOKEN)
      # text tokens: standard text tokenization
      if text is not None:
          text_tokens = text_tokenizer.encode(text)
          tokens.extend(text_tokens)
      # audio tokens (optional)
      if audio is not None:
          tokens.append(AUDIO_START_TOKEN)
          audio_tokens = audio_encoder.encode_audio_to_tokens(audio)
          tokens.extend(audio_tokens)
          tokens.append(AUDIO_END_TOKEN)
      return tokens

4. 多模态数据格式标准化 交互式图文数据（Interleaved Image-Text）的常见格式： MMC4 格式（Multimodal C4，Allen AI）： { "text_list": ["A photo of ", null, " and a ", null, "."], "image_list": [, ], "image_index": [1, 3], "similarity_matrix": [[0.85, 0.23], [0.12, 0.78]] } OBELICS 格式（HuggingFace）： { "documents": [ {"content_type": "text", "text": "Here is a picture of a cat:"}, {"content_type": "image", "image": ""}, {"content_type": "text", "text": "The cat is sitting on a sofa."} ] } 多模态数据通常以 WebDataset 或 MosaicML MDS 格式存储，支持高效的随机访问和多模态同时加载。
5. 多分辨率图像处理 多分辨率处理是多模态模型中的关键技术突破（LLaVA-1.6 / LLaVA-NeXT）： def multi_resolution_process(image, base_size=336, max_num_tiles=4): """ Split large image into multiple tiles, each tile encoded independently Reference: LLaVA-1.6 AnyRes strategy """ w, h = image.size

     # calculate tile count (maintain aspect ratio, not exceeding max_num_tiles)
     tiles_w = min(max(1, w // base_size), max_num_tiles)
     tiles_h = min(max(1, h // base_size), max_num_tiles)
     tiles = []
     for i in range(tiles_h):
         for j in range(tiles_w):
             tile = image.crop((
                 j * base_size, i * base_size,

(j + 1) * base_size, (i + 1) * base_size ))

             tiles.append(tile)
     # add global thumbnail (resize full image to base_size)
     global_view = image.resize((base_size, base_size))
     tiles.insert(0, global_view)
     return tiles

多分辨率策略的代价是计算量随 tile 数线性增长（一个 6-tile 图像的处理量约为基础图像的 6 倍），但大幅提升对高分辨 率图像中细节的理解能力。 6) 多模态管线性能瓶颈 多模态预处理管线的主要瓶颈是图像解码和编码（占预处理时间的 60-80%），而非文本 tokenization。优化策略：

1. GPU 加速解码：使用 NVIDIA DALI 进行 GPU JPEG 解码和图像变换
2. 预提取嵌入：对于使用固定视觉编码器（如 CLIP）的场景，可一次性将图像编码为嵌入向量，训练时直接加载向量 （而非原始图像）
3. WebDataset 流式加载：tar 格式的顺序读取避免随机 I/O 带来的延迟
4. 多分辨率并行处理：将不同分辨率 tile 的处理分配到不同 GPU/CPU 核心

13.7.2 视频数据管线

视频数据容量远超文本：一段 10 秒 1080p 视频（30fps）300 帧约 500 MB，而模型训练通常只需 8-16 帧。核心矛盾是 原始数据大、模型消费小，管线必须先抽帧/降采样，把几 TB 的原始视频转化为几 GB 的训练样本。预处理（解码、抽 帧、字幕对齐、清晰度过滤）通常在离线批量完成，产出压缩后的训练样本（如 16 帧 × 分辨率 × RGB），训练时直接消 费。

1. 帧采样与音视频对齐 帧采样策略：

 # Requires opencv-python and numpy
 import cv2
 def sample_frames(video_path, num_frames=8, fps_target=1):
     """

Uniformly sample num_frames from the video Strategy: FPS downsampling + uniform interval sampling

     """
     cap = cv2.VideoCapture(video_path)
     total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
     original_fps = cap.get(cv2.CAP_PROP_FPS)
     if total_frames < num_frames:
         return None # video too short, discard
     # FPS downsampling step
     step = int(original_fps / fps_target)
     # select num_frames uniformly spaced frame indices
     indices = np.linspace(0, total_frames - 1, num_frames, dtype=int)
     indices = indices[::step] # further filter by FPS
     frames = []
     for idx in indices:
         cap.set(cv2.CAP_PROP_POS_FRAMES, idx)

ret, frame = cap.read() if not ret: break

         frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
         frames.append(frame)
     cap.release()
     return frames

视频-音频对齐：使用 Whisper 提取音频轨道并转录为文本，用于视频片段与文本的时间对齐：

 # Requires openai-whisper
 import whisper
 model = whisper.load_model("large-v3")
 def extract_audio_text(video_path):
     result = model.transcribe(video_path)
     segments = result["segments"]
     # each segment has (start_time, end_time, text)
     # used for text alignment of video segments
     return segments

2. 解码与抽帧并行 解码是视频管线的计算瓶颈。原始视频需 H.264/HEVC 解码才能取帧，解码 CPU 开销大，处理链路： Video file -> HW decode (GPU/NVDEC/VPU) -> frame sampling (by fps/scene) -> preprocessing -> encode to training sample s •硬件解码：用 GPU 硬件解码器（NVDEC）或专用解码卡批量解码，CPU 软解太慢（千万小时级不可行） •抽帧策略：均匀抽帧（按时间间隔）或场景抽帧（按内容变化），影响训练样本的时间覆盖 •并行化：解码任务按视频文件分片并行（每个 worker 处理一批视频），解码结果落盘为训练样本
3. 流式加载与多级缓存 训练时直接读大视频文件会频繁 IO 阻塞。流式加载（streaming pipeline）把“读盘 → 解码 → 预处理 → 上 GPU”流水 线化，配合多级缓存消除瓶颈： •多级缓存：SSD 缓存（热样本）→ 内存缓存（当批样本）→ 显存（当前 batch）。样本按访问频率分层，最热的驻留内 存，避免重复读盘 •预取与重叠：当前 batch 训练时，下一 batch 已在解码/预处理，消除 IO 等待（双缓冲思想） •缓存预处理结果：对重复访问的样本缓存预处理结果（如已抽帧的图），避免每次重新解码；缓存命中率高时显著降低 CPU 开销
4. 视频索引 千万小时级视频需要内容级索引才能高效检索与去重： •帧级索引：视频到帧的时间戳映射（秒级定位） •内容索引：视频元数据（时长、分辨率、清晰度、内容标签）入库，支持按内容筛选（过滤低质/重复视频） •去重索引：视频级去重（近似哈希）+ 帧级去重（感知哈希），防止重复样本浪费算力
5. 图文视频混合管线 多模态训练常混合图文与视频样本，管线需支持异构 batch： •图文样本小（几 KB-几 MB），视频样本大（处理后仍数百 KB），混合采样需按样本大小动态平衡 •两种模态的预处理链路不同（图直接加载，视频需解码），worker 池按模态分组避免互相阻塞 •动态 shape：图像/视频帧分辨率不一，dataloader 产出动态 shape batch，模型侧配合动态 padding 或按 bucket 分 组

13.7.3 数据飞轮与在线学习

数据飞轮（Data Flywheel）是 AI 系统从部署中持续改进的自我增强机制：模型的部署产生用户交互数据，这些数据被收 集、标注和用于重新训练模型，改进后的模型又产生更好的用户体验和更多的高质量交互数据。这个循环是 ChatGPT、 Copilot 等产品的核心质量迭代引擎。

1. 数据飞轮架构 数据飞轮循环如图13-6所示。 Annotation Data Collection Data Annotation User interaction Query/Response/Feedbac Manual/Automated New data Model Deployment k Data Quality Monitoring Model Retraining Inference Service Analysis Distribution Shift Detectio Guidance SFT/RLHF/DPO n New model version 图13-6 数据飞轮循环 飞轮的每个周期包含四个阶段：

1. 交互收集：记录用户查询、模型响应、用户反馈（点赞/踩、编辑、重新生成）
2. 数据标注：将原始交互转化为可训练的数据点（偏好对、改进响应、拒绝响应）
3. 模型改进：使用新数据通过 SFT（监督微调）、RLHF 或 DPO 进行模型更新
4. 部署验证：新模型通过 A/B 测试与旧模型对比，确认改进后推全

2. RLHF 数据收集 RLHF（Reinforcement Learning from Human Feedback）是 ChatGPT 成功的关键技术，其数据收集流程如下： 阶段 1（SFT 数据收集）：人类标注者撰写高质量的对话示范（demonstrations），覆盖不同难度、领域和风格的查询-响 应对，典型规模为 10K-50K 对。 阶段 2（偏好数据收集）：对于每个查询，模型生成多个候选响应（如 4 个），人类标注者对这些响应进行排名（ranking） 或成对比较（pairwise comparison），典型的收集规模为 100K-500K 对比对。

preference data standard format

preference_data = { "prompt": "Explain the concept of neural networks to a 10-year-old.", "chosen": "Think of a neural network like a team of tiny...", # human-preferred response "rejected": "A neural network is a computational graph...", # rejected response "metadata": {"annotator_id": "A123", "quality_score": 4.5} } 标注质量控制： •多数投票（Majority Voting）：每个对比对由 3-5 个标注者独立评估，取多数意见 •标注者一致性检查：定期插入黄金标准问题（已知正确答案），评估标注者的一致性 •对抗性探测：插入故意错误的标注对，检测标注者是否认真 3) DPO 直接偏好优化 DPO（Direct Preference Optimization, 2023, Stanford）简化了 RLHF 的复杂度，直接使用偏好数据优化模型，不需要 训练单独的奖励模型（Reward Model）和执行 PPO 强化学习。DPO 损失函数为： πθ (yw ∣x) π (y ∣x) LDPO = − log σ (β log − β log θ l ) πref (yw ∣x) πref (yl ∣x) 其中 y 是 chosen 响应，y 是 rejected 响应，π 是正在优化的模型，π 是参考模型（通常是 SFT 模型），β 控制偏离 参考模型的程度。 w l θ ref

 # DPO training loop (simplified)
 import torch.nn.functional as F
 def dpo_loss(model, ref_model, batch, beta=0.1):
     # batch: prompt, chosen_response, rejected_response
     chosen_logps = model(batch["prompt"], batch["chosen"]).log_probs
     rejected_logps = model(batch["prompt"], batch["rejected"]).log_probs
     ref_chosen_logps = ref_model(batch["prompt"], batch["chosen"]).log_probs
     ref_rejected_logps = ref_model(batch["prompt"], batch["rejected"]).log_probs
     chosen_reward = beta * (chosen_logps - ref_chosen_logps)
     rejected_reward = beta * (rejected_logps - ref_rejected_logps)
     loss = -F.logsigmoid(chosen_reward - rejected_reward).mean()
     return loss

DPO 的优势：训练稳定（单一损失函数，无需对奖励模型进行 PPO 优化），硬件需求低（不需要同时维护 Actor/Critic/Reward 模型 + 参考模型），这使得偏好训练对中小团队也变得可行。 4) LIMA 与迭代 DPO LIMA（Zhou et al., 2023）发现仅用 1,000 个精心策划的 SFT 样例（覆盖约 300 个任务类型），在 65B LLaMA 上微调即 可达到接近 GPT-4 早期的对话质量，在 300 个人类评估测试集上胜率 43% vs DaVinci003（Alpaca-65B 仅 19%，尽管使 用了 52K 样例）。这证明数据质量比数据量重要 10-50 倍。Iterative DPO 方面，Llama 3 是首个公开完整流水线的生产级 模型：每轮 DPO 使用约 100 万偏好对，模型自标注生成下一轮数据，经过 5 轮迭代 DPO 后 AlpacaEval 2.0 胜率显著提 升。此类自动评估与奖励坍塌的缓解需要引入人类标注锚点作为校准基准。 5) 宪法 AI Anthropic 提出的宪法 AI 是一种自动化数据飞轮方法，不依赖人类标注者，而是使用 AI 自身来评估和改进自己的输出。 工作流程：

1. 生成与修正：模型生成初始响应，然后根据“宪法”（一组原则，如“不要提供导致伤害的信息”）自我修正
2. AI 反馈：另一个模型评估原始响应和修正后响应，生成偏好对
3. 训练：使用这些 AI 生成的偏好对训练模型（通过 RLHF 或 DPO） 宪法 AI 大大降低了人类标注成本和等待时间，将数据飞轮从“周/月”级别加速到“小时/天”级别。Anthropic Claude 的多次迭代均受益于此方法。

6. 在线学习与流式数据 在线学习（Online Learning）允许模型在处理新数据时增量更新参数，而非需要完整的批量重训练。LLM 的在线学习面 临独特挑战： •灾难性遗忘：新数据驱动模型适应新分布时，可能丢失已学到的能力 •数据分布偏移：用户查询随时间变化（季节、热点事件），模型需持续适应 Replay Buffer 是缓解灾难性遗忘的常用技术：

 class ReplayBuffer:
     def __init__(self, capacity=100000):
         self.buffer = deque(maxlen=capacity)
         self.capacity = capacity
     def add(self, sample):
         self.buffer.append(sample)
     def sample(self, batch_size, current_batch):
         # mix current data + historical data (e.g. 70% new + 30% old)
         replay_batch = random.sample(self.buffer, min(len(self.buffer), batch_size // 3))
         combined = current_batch + replay_batch
         random.shuffle(combined)
         return combined

在线 DPO（迭代 DPO）在 Llama 3 的生产中已被验证：在每轮 DPO 训练后，使用当前最优模型生成新的候选响应用于 下一轮偏好数据收集，形成“模型改进 → 更多数据 → 模型改进”的闭环。 7) A/B 质量验证 数据飞轮的效果必须通过严格的 A/B 测试来量化。典型指标：

 # A/B testing evaluation
 def ab_test(control_model, treatment_model, test_queries, n=1000):
     metrics = {

"win_rate": 0, # proportion of users preferring treatment "edit_distance": 0, # frequency of user editing responses (lower is better) "regeneration_rate": 0, # proportion of user regenerations "response_length": 0, # average response length

     }
     for query in test_queries[:n]:
         response_a = control_model.generate(query)
         response_b = treatment_model.generate(query)
         # present to user (blind evaluation), record selection

... return metrics 数据飞轮的关键非技术挑战是隐私和合规：用户交互数据可能包含敏感信息，必须经过脱敏处理后才可用于模型训练。 GDPR 的被遗忘权意味着用户数据可能需要从训练集中完全移除，这对数据管线的审计和重放能力提出了严格要求。 8) 工具与生态 数据飞轮工程化工具： •LangSmith（LangChain）：追踪 LLM 调用链，收集用户反馈，构建评估数据集 •Weights & Biases：实验追踪 + 模型版本管理 •Label Studio：开源数据标注平台，支持 RLHF 偏好标注工作流 •Argilla：开源数据标注和反馈收集平台，专为 NLP/LLM 优化 •Custom Solutions：自行开发的反馈收集 API，直接集成到推理网关中 数据飞轮的工程本质是将“用户反馈 → 模型改进”的周期从手动的数周加速到自动化的数天或数小时，是保持 LLM 产品 持续竞争力的关键基础设施。

13.7.4 后训练数据验收与标注

后训练（SFT、偏好优化、RL）数据与预训练数据的性质不同：规模更小、质量要求更高、正确性有明确标准。相应地， 数据管线在预训练之外需要一套独立的验收与标注体系，确保进入训练环节的样本满足指令遵循、无害性与任务正确性约 束。

1. 验收标准设计 后训练样本的验收维度因用途而异。指令微调数据关注指令-响应对齐（响应是否直接回答指令、是否遵循格式约束）；偏 好数据关注成对比较的判别一致性（正负样本是否存在可判定的优劣关系）；RL 数据关注结果可验证性（数学、代码类任 务可自动判定对错）。验收标准须写成可操作的检查项，而非抽象描述，才能支撑规模化质检。
2. 验收管线 验收管线通常分三层递进。规则层用正则与启发式过滤格式错误、重复样本、涉敏内容；模型层用强模型对响应打分或做 分类判别，覆盖规则层无法判定的语义质量问题；人工层对模型判定低置信度的样本抽检复核。三层结合在吞吐与准确率 之间取得平衡，人工只处理机器难以裁决的边界样本。
3. 标注体系设计 标注体系包括标注规范、任务分解与质量监控三部分。标注规范把验收标准转化为逐条可执行的标注指引，并配以典型正 反例；复杂任务（如多轮对话、代码审查）按子任务分解给不同专长的标注小组；质量监控通过标注一致性（同一样本多 人标注的吻合率）与抽检打回率跟踪标注质量，形成标注质量的闭环反馈。
4. 面向后训练的合成数据 合成数据是扩充后训练数据的重要手段。以种子指令为基础，用强模型批量生成响应，再经验收管线筛选。合成数据的价 值取决于生成-筛选的闭环质量，而非生成量本身：低质量生成淹没在高吞吐下会拉低训练数据整体质量。面向后训练的 合成管线与预训练合成数据管线复用底层存储，但验收标准显著更严，两者在数据飞轮中共享样本生命周期管理。

13.8 PB 级数据管线实战

以下提供一个完整的 PB 级语料清洗与分词管线的工程实现。从 Common Crawl 原始 WET 文件出发，经过提取、过滤、 去重、分词、打包，最终产出可直接用于 LLM 预训练的 MDS（Mosaic Data Shard）格式数据集。

13.8.1 管线架构

PB 级语料处理管线的六个阶段如图13-7所示。 Common Crawl WET ~800TB raw data Phase 1: Text Extraction WARC/WET ▶ Plain Text Phase 2: Quality Filter Lang detection + Length + PPL + Toxicity Phase 3: Deduplication MinHash LSH near-dedup Phase 4: Tokenization BPE/SentencePiece Tokeni zation Phase 5: Sequence Packin g Pack to seq_len=8192 Phase 6: Shard Output MDS format, 1024 shards Training Ready Data ~3TB tokenized 图13-7 PB级语料处理管线 计算资源需求：64 节点集群，每节点 64 CPU 核心 + 256 GB RAM + 4×NVMe SSD，总 CPU 核心数 4096。

1. 文本提取

 # extract_text.py: extract plain text from WET files into JSONL shards
 # Requires Python 3.10+, warcio, gzip
 import gzip
 import warcio
 from tqdm import tqdm
 import os
 def extract_wet_files(wet_paths, output_dir, max_docs_per_shard=100000):
     """

Extract plain text from WET files and save as JSONL shards WET format: WARC/1.0 container, each record contains: URI, Content-Type, text content

     """
     shard_idx = 0
     doc_count = 0
     output_file = None
     for wet_path in wet_paths:

with gzip.open(wet_path, 'rb') as stream: for record in warcio.ArchiveIterator(stream): if record.rec_type != 'conversion': continue

                 uri = record.rec_headers.get_header('WARC-Target-URI')
                 content = record.content_stream().read().decode('utf-8', errors='replace')
                 if len(content) < 200:   # minimum length threshold

continue

                 # initialize shard file (create on demand)
                 if doc_count % max_docs_per_shard == 0:
                     if output_file:
                         output_file.close()
                     output_file = open(

f"{output_dir}/extracted-{shard_idx:05d}.jsonl", "w" ) shard_idx += 1 output_file.write(json.dumps({ "uri": uri, "text": content, "length": len(content), }) + "\n")

                 doc_count += 1
     if output_file:
         output_file.close()
     print(f"Extracted {doc_count} documents into {shard_idx} shards")

预期产出：从 800 TB WET 文件中提取约 500 亿个文档（约 100 TB 纯文本）。 2) 质量过滤

 # quality_filter.py
 # Requires Ray 2.x and pycld2
 import ray
 import pycld2
 import json
 ray.init(address="auto")   # connect to Ray cluster

@ray.remote(num_cpus=1) def filter_shard(shard_path, output_path): """ Process quality filtering for a single shard:

1. Language detection (keep only English/configurable)
2. Length filter (100-100000 characters)
3. Basic stats filter (alpha ratio, line dedup) """ filtered = [] with open(shard_path) as f:

         for line in f:
             doc = json.loads(line)
             text = doc["text"]
             # language detection

try: reliable, _, details = pycld2.detect(text) if not reliable or details[0][1] != 'en': continue except Exception: continue # length filter if len(text) < 100 or len(text) > 100000: continue

             # alpha ratio filter
             alpha_ratio = sum(c.isalpha() for c in text) / len(text)
             if alpha_ratio < 0.6:

continue

             # line dedup filter
             lines = text.split('\n')
             lines = [l for l in lines if len(l) > 10]
             if len(lines) < 2:

continue unique_ratio = len(set(lines)) / len(lines) if unique_ratio < 0.5: continue filtered.append(doc) with open(output_path, "w") as f:

         for doc in filtered:
             f.write(json.dumps(doc) + "\n")
     return len(filtered)
 # process all shards in parallel
 shard_paths = sorted(glob.glob("extracted-*.jsonl"))
 futures = [
     filter_shard.remote(

path, path.replace("extracted-", "filtered-") ) for path in shard_paths ] counts = ray.get(futures) print(f"Filtered: {sum(counts)} documents retained from {len(shard_paths)} shards") 过滤后保留率约 15-25%，从 500 亿文档中保留约 75-125 亿文档。 3) MinHash 去重

 # deduplicate.py
 # Requires Ray 2.x and datasketch
 from datasketch import MinHash, MinHashLSH
 import ray
 import hashlib

@ray.remote(num_cpus=2)

 class DedupWorker:
     def __init__(self, num_perm=128, threshold=0.8):
         self.num_perm = num_perm
         self.threshold = threshold
         self.lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)
         self.minhashes = {}
     def insert_document(self, doc_id, text):
         m = MinHash(num_perm=self.num_perm)
         # create MinHash signature using 5-gram
         for i in range(len(text) - 4):
             ngram = text[i:i+5]
             m.update(ngram.encode('utf8'))
         # query near duplicates
         result = self.lsh.query(m)
         self.lsh.insert(doc_id, m)
         self.minhashes[doc_id] = m
         return result   # return list of near-duplicate IDs found
     def get_unique_ids(self):
         # return all unique document IDs in LSH
         return list(self.minhashes.keys())
 # distributed dedup execution
 def distributed_dedup(input_dir, output_dir, num_workers=32):
     workers = [DedupWorker.remote() for _ in range(num_workers)]
     # assign document hash to specific worker (based on doc_id hash)
     shard_files = sorted(glob.glob(f"{input_dir}/filtered-*.jsonl"))
     unique_count = 0
     for shard_file in shard_files:

with open(shard_file) as f:

             for line in f:
                 doc = json.loads(line)
                 doc_hash = hashlib.md5(doc["text"].encode()).hexdigest()
                 worker_idx = int(doc_hash, 16) % num_workers
                 # insert and get near duplicates
                 ray.get(workers[worker_idx].insert_document.remote(doc_hash, doc["text"]))
     # collect all unique IDs
     all_unique = set()
     for worker in workers:
         worker_unique = ray.get(worker.get_unique_ids.remote())
         all_unique.update(worker_unique)
     print(f"Unique documents: {len(all_unique)} (dedup ratio: {len(all_unique)/unique_count:.2%})")
     return all_unique

MinHash LSH 去重后保留率约 60-80%（取决于数据源的原始重复率）。对于 Common Crawl，预计保留 45-100 亿唯一 文档。 4) 分词与序列封包

 # tokenize_and_pack.py
 # Requires Python 3.10+, transformers and numpy
 from transformers import AutoTokenizer
 import numpy as np
 def tokenize_and_pack(input_file, output_file, tokenizer, seq_len=8192):
     """

Tokenize text and pack multiple document token sequences to fixed length Use EOS token to separate different documents

     """
     tokenizer = AutoTokenizer.from_pretrained("meta-llama/Meta-Llama-3-70B")
     eos_token_id = tokenizer.eos_token_id
     buffer = [] # token buffer
     packed_sequences = []

with open(input_file) as f:

         for line in f:
             doc = json.loads(line)
             tokens = tokenizer.encode(doc["text"], add_special_tokens=False)
             tokens.append(eos_token_id) # document separator
             buffer.extend(tokens)
             # when buffer accumulates enough tokens, pack into complete sequence
             while len(buffer) >= seq_len:
                 seq = buffer[:seq_len]
                 packed_sequences.append(seq)
                 buffer = buffer[seq_len:]
     # save packed sequences as numpy array
     data = np.array(packed_sequences, dtype=np.int32)
     np.save(output_file, data)
     return len(packed_sequences)

对于 3 TB tokenized 数据 + seq_len=8192 的配置：3 TB / (8192 × 4 bytes) ≈ 91,552,734 条序列。 5) MDS 格式输出

 # convert_to_mds.py
 # Requires streaming (MosaicML) and numpy
 from streaming import MDSWriter
 import numpy as np
 def convert_to_mds(input_dir, output_dir, num_shards=1024):
     """

Convert tokenized numpy sequences to MDS shard format Each shard contains {'input_ids': np.array, 'labels': np.array}

     """
     input_files = sorted(glob.glob(f"{input_dir}/*.npy"))
     shard_writers = [

MDSWriter(

             out=f"{output_dir}/shard-{i:04d}",
             columns={"input_ids": "ndarray:int32", "labels": "ndarray:int32"},
             compression="zstd:7",
             size_limit=1 << 29, # 512 MB per shard

) for i in range(num_shards) ]

     for i, file in enumerate(input_files):
         data = np.load(file)
         shard_idx = i % num_shards
         for j in range(data.shape[0]):
             seq = data[j]
             sample = {

"input_ids": seq[:-1], "labels": seq[1:], # language model next token prediction } shard_writers[shard_idx].write(sample)

13.8.2 管线编排与监控

     for writer in shard_writers:
         writer.close()
     print(f"Written {num_shards} MDS shards to {output_dir}")

六个阶段的串行编排通过 shell 脚本完成，每阶段独立日志记录：

 #!/bin/bash
 # run_full_pipeline.sh
 set -e
 PREFIX="llama-70b-data"
 WET_PATH="s3://commoncrawl/cc-index/collections/CC-MAIN-2024-*"
 OUTPUT_BASE="s3://my-dataset/processed"

echo "=== Phase 1: Extraction ===" python extract_text.py \

   --wet-paths "${WET_PATH}" \
   --output-dir "${OUTPUT_BASE}/01-extracted/" \
   --max-docs-per-shard 100000 \

2>&1 | tee logs/01_extract.log echo "=== Phase 2: Quality Filtering ===" python quality_filter.py \

   --input-dir "${OUTPUT_BASE}/01-extracted/" \
   --output-dir "${OUTPUT_BASE}/02-filtered/" \
   --language en \
   --min-length 100 --max-length 100000 \

2>&1 | tee logs/02_filter.log echo "=== Phase 3: Deduplication ===" python deduplicate.py \

   --input-dir "${OUTPUT_BASE}/02-filtered/" \
   --output-dir "${OUTPUT_BASE}/03-deduped/" \
   --num-perm 128 --threshold 0.8 \

2>&1 | tee logs/03_dedup.log echo "=== Phase 4: Tokenization ===" python tokenize_and_pack.py \

   --input-dir "${OUTPUT_BASE}/03-deduped/" \
   --output-dir "${OUTPUT_BASE}/04-tokenized/" \
   --tokenizer meta-llama/Meta-Llama-3-70B \
   --seq-len 8192 \

2>&1 | tee logs/04_tokenize.log echo "=== Phase 5: MDS Conversion ===" python convert_to_mds.py \

   --input-dir "${OUTPUT_BASE}/04-tokenized/" \
   --output-dir "${OUTPUT_BASE}/05-mds/" \
   --num-shards 1024 \

2>&1 | tee logs/05_mds.log echo "=== Pipeline Complete ==="

13.8.3 性能优化要点

1. I/O 优化：所有中间结果使用 Parquet 格式（列式 + 压缩），避免 JSONL 的文本解析开销。Parquet 读取速度比 JSONL 快 5-10×
2. 内存优化：Ray Data 的 map_batches 指定 batch_size=10000 ，平衡内存和吞吐
3. 网络优化：MinHash LSH 的去重阶段，将 doc_id 哈希到 worker 可以避免全量数据 shuffle，减少 90% 的网络传输
4. 存储分层：中间结果写入 NVMe，最终 MDS 输出到 S3

13.8.4 管线耗时与成本估计

在 64 节点 × 64 CPU 核心 = 4096 核心集群上的预期耗时如表13-12所示。 表13-12 PB管线阶段耗时估计 阶段 输入量 输出量 预计耗时 CPU 利用率

1. 文本提取 800 TB WET 100 TB 文本 6-8 小时 80% (I/O-bound) 阶段 输入量 输出量 预计耗时 CPU 利用率
2. 质量过滤 100 TB 文本 25 TB 过滤后 10-12 小时 70% (CPU-bound)
3. MinHash 去重 25 TB 18 TB 去重后 12-16 小时 60% (网络+CPU 混合)
4. 分词+封包 18 TB 3 TB Tokenized 20-24 小时 90% (CPU-bound)
5. MDS 转换 3 TB 3 TB MDS 2-3 小时 40% (I/O-bound) 总计 800 TB 3 TB 50-63 小时 — 总计算成本（按 AWS EC2 c6i.16xlarge spot 实例 $0.50/hr 计算）：64 × 60 × $0.50 ≈ $1,920。

13.8.5 管线验证

处理完成后的验证步骤：

 # verification script
 # Requires streaming (MosaicML) and transformers
 from streaming import StreamingDataset
 # load and verify MDS dataset
 dataset = StreamingDataset(local="/tmp/test", remote="./data/tokenized/")
 print(f"Total samples: {len(dataset)}")
 sample = dataset[0]
 print(f"Input shape: {sample['input_ids'].shape}")
 print(f"Label shape: {sample['labels'].shape}")
 # decode and confirm data quality
 tokenizer = AutoTokenizer.from_pretrained("meta-llama/Meta-Llama-3-70B")
 decoded = tokenizer.decode(sample['input_ids'][:100])
 print(f"Sample text: {decoded[:200]}...")
 # count sequence length distribution
 lengths = [dataset[i]['input_ids'].shape[0] for i in range(10000)]
 print(f"Length stats: min={min(lengths)}, max={max(lengths)}, mean={np.mean(lengths)}")

assert max(lengths) <= 8192, "Sequence length exceeds limit" 最终输出：1024 个 MDS 分片，总计约 3 TB，包含约 9100 万条序列，可直接用于 Llama-70B 的大规模分布式预训练。