章节导航 第 28 章 / 共 30 章
第 28 章 篇六：基础设施与质量闭环

数据工程基础设施与工具链

支撑数据工程的存储、调度、分布式处理与工具链。

基础设施 工具链 数据管道 DataFlow Agentic数据工程

“好的数据工程不仅在于’做什么’，还在于’用什么做’——基础设施决定了你能做到的上限。”

你可以有最好的数据清洗算法，但如果无法在合理的时间和成本内处理 10PB 的数据，一切都是空谈。这一章是”工具篇”——从框架选择到分布式实现，从版本管理到自动化管线，帮你搭建一个生产级的数据工程基础设施。

28.1 数据处理框架

主流框架对比

框架开发方核心特点适用规模语言许可
datatroveHuggingFaceFineWeb 的实际框架，模块化、可扩展PB 级PythonApache 2.0
NeMo-CuratorNVIDIAGPU 加速，多模态（文本/图像/视频/音频），企业级特性PB 级PythonApache 2.0
dolmaAI2OLMo 数据管线，生产导向（1.0+），Rust+Python 混合PB 级Rust + PythonApache 2.0
RedPajama-DataTogether AI完整参考实现，易读TB 级PythonApache 2.0
cc_netMetaCCNet 管线，专注 CC 处理PB 级PythonMIT
自建方案-完全可控，深度定制取决于工程投入--

与上述框架定位不同的一类：DataFlow（北大 OpenDCAI，Apache-2.0）。上表五个框架的处理对象都是预训练语料——从 Common Crawl 的 WARC 字节到去重后的 token 流。DataFlow 处理的是另一段：把 PDF、纯文本、低质量 QA 变成可用于 SFT / RL / RAG 的监督数据，即篇三、篇四的对象。

抽象是 Pipeline → Operator → Prompt 三层，算子按 Generate / Evaluate / Filter / Refine 四类组织，serving 层统一封装 OpenAI API / vLLM / 本地推理，分布式走 Ray。据其技术报告（arXiv:2512.16676，2025-12-18 提交）：框架含近 200 个可复用算子与六条领域通用流水线（文本、数学推理、代码、Text-to-SQL、Agentic RAG、大规模知识抽取），并配 DataFlow-Agent 把自然语言需求经「算子合成 → 管线规划 → 迭代验证」翻译成可执行管线。六个用例上的报告结果为：Text-to-SQL 较 SynSQL +3% 执行准确率、代码基准平均 +7%、MATH/GSM8K/AIME +1–3 分；且 10K 条 DataFlow 数据的效果超过 1M 条 Infinity-Instruct。该报告已被 ACM SIGMOD 2027 接收（2026-09-12 公布）。

⚠️ 不要把 DataFlow 当成 datatrove / NeMo-Curator 的替代品：它不做 TB–PB 级的 MinHash 去重，也不覆盖预训练的语种识别、安全过滤、PII 脱敏链路。正确用法是分层——预训练语料清洗仍走 datatrove / NeMo-Curator（见上文决策树），DataFlow 承接清洗之后「把干净文本变成训练样本」的那一段。

两处口径差异，引用时勿混用：arXiv 摘要（一手）写「nearly 200 reusable operators」，指框架算子总数；仓库 README 另给「10+ 核心算子 + 100+ 流水线专用算子」的细分口径。二者统计对象不同，本书正文统一采用摘要口径。

框架选择决策树

先问：你有没有自有（或可长期租用的）集群？        ← 这条与数据量正交，先看它
├── 没有 → datatrove JobsPipelineExecutor（0.10.0+，experimental）
│          把 pipeline 提交到 Hugging Face Jobs 执行，零本地 GPU
│          tasks 分片 + depends 阶段依赖 + skip_completed 断点续跑 + max_retries
│          按 Job 指定硬件 flavor（默认 cpu-basic，可换 GPU）
│          成本锚点：3.6 万张模型/数据卡 → 27B 教师 → 35,837 条摘要 ≈ $15.58（公开实测）
└── 有 → 继续看下面两棵树

你的数据量有多大？
├── < 100GB → 简单 Python 脚本就够了（pandas/datasets）
├── 100GB - 1TB → datatrove 单机模式 / RedPajama
├── 1TB - 100TB → datatrove 分布式（LocalPipelineExecutor / SlurmPipelineExecutor / RayPipelineExecutor）
│   ├── 有 GPU 集群 → 考虑 NeMo-Curator（去重和过滤加速 5-10x）
│   └── 只有 CPU → datatrove + Spark/Dask
└── > 100TB → 自建 + datatrove/dolma 混合
    └── 必须考虑：存储架构、网络带宽、容错机制

你的团队情况？
├── 熟悉 HuggingFace 生态 → datatrove
├── 有 NVIDIA 合作 → NeMo-Curator
├── 重视可复现性 → dolma（AI2 文档最完善）
└── 需要快速原型 → RedPajama

执行器怎么选？（datatrove 0.10.x 起是四选一，不再是三选一）
├── 本机多核 → LocalPipelineExecutor
├── Slurm 集群 → SlurmPipelineExecutor
├── Ray 集群 → RayPipelineExecutor
└── 无集群 / 想按次付费 → JobsPipelineExecutor（提交到 HF Jobs，见决策树最上面那条）

你的数据包含哪些模态？
├── 纯文本 → datatrove / dolma（最成熟）
├── 多模态（文本+图像/视频/音频） → NeMo-Curator（2026 年已支持多模态策展）
│   ├── 跨模态去重（语义相同但像素不同的图片）
│   ├── 视觉-语言对齐质量度量
│   └── 多模态 Scaling Law 不同于纯文本
└── Agent 轨迹（工具调用/多步推理/环境交互） → 需要自建方案
    └── 这是新兴方向，现有框架尚未覆盖

datatrove 实战

datatrove 是当前社区最活跃的数据处理框架，FineWeb 2 万亿 token 数据集就是用它构建的。

"""
datatrove 实战：构建完整的预训练数据处理管线
从 Common Crawl WARC 到干净的训练数据
"""

# pip install "datatrove[all]" "xxhash<4"
#
# ⚠️ 依赖钉死警告：xxhash 4.0.0 会破坏所有去重管线。
# datatrove 官方在 0.10.0 里显式把 xxhash 钉到 <4（PR #507，原话 "4.0.0 breaks all dedup
# pipelines"），注意该约束只挂在 processing extra 上，装 datatrove 主包时不会自动带上。
# 这类变更属于"静默破坏"——不抛异常，只悄悄改掉去重结果。因此 MinHash 类管线在升级
# 任何依赖之后，都必须比对产物指纹（与第 30 章 §30.10「加速前先固化 CPU 基线产物指纹」
# 是同一条纪律）。

from datatrove.pipeline.readers import WarcReader
from datatrove.pipeline.filters import (
    URLFilter,
    LanguageFilter,
    GopherQualityFilter,
    GopherRepetitionFilter,
    C4QualityFilter,
)
from datatrove.pipeline.dedup import (
    MinhashDedupSignature,
    MinhashDedupBuckets,
    MinhashDedupCluster,
    MinhashDedupFilter,
)
from datatrove.pipeline.writers import JsonlWriter
from datatrove.executor import LocalPipelineExecutor

# ====== 阶段 1：文本提取 + 基础过滤 ======
stage1_executor = LocalPipelineExecutor(
    pipeline=[
        # 从 WARC 文件读取（使用 trafilatura 提取文本）
        WarcReader(
            data_folder="s3://commoncrawl/crawl-data/CC-MAIN-2024-18/segments/",
            glob_pattern="*/warc/*",
        ),
        # URL 过滤（去除成人网站、广告网站等）
        URLFilter(
            exclusion_writer=JsonlWriter("output/stage1/excluded_urls/"),
        ),
        # 语言识别过滤（只保留中文和英文）
        LanguageFilter(
            languages=["zh", "en"],
            language_threshold=0.65,  # 置信度阈值
        ),
        # Gopher 质量过滤（Google DeepMind 的规则集）
        GopherQualityFilter(
            min_doc_words=50,           # 最少 50 词
            max_doc_words=100000,       # 最多 10 万词
            min_avg_word_length=3,      # 平均词长 ≥ 3 字符
            max_avg_word_length=10,     # 平均词长 ≤ 10 字符
            min_stop_words_ratio=0.06,  # 停用词比例下限
        ),
        # Gopher 重复检测（行级 / 段落级去重）
        GopherRepetitionFilter(
            dup_line_frac=0.3,       # 重复行比例上限
            dup_para_frac=0.3,       # 重复段落比例上限
            dup_line_char_frac=0.2,  # 重复行字符比例上限
            top_n_grams=(2, 3, 4),   # 检测 top n-gram 重复
            top_n_grams_frac=(0.2, 0.18, 0.16),
        ),
        # 写入中间结果
        JsonlWriter("output/stage1/clean/"),
    ],
    tasks=128,  # 并行任务数
    workers=32, # 工作进程数
)

# ====== 阶段 2：MinHash 模糊去重 ======
# 分 4 步执行（datatrove 的设计）

# Step 2a: 计算 MinHash 签名
stage2a = LocalPipelineExecutor(
    pipeline=[
        MinhashDedupSignature(
            input_folder="output/stage1/clean/",
            output_folder="output/stage2/signatures/",
            n_sentences=5,       # 每个 shingle 包含 5 个句子
            num_hashes=128,      # MinHash 签名维度
        ),
    ],
    tasks=128,
)

# Step 2b: LSH 分桶
stage2b = LocalPipelineExecutor(
    pipeline=[
        MinhashDedupBuckets(
            input_folder="output/stage2/signatures/",
            output_folder="output/stage2/buckets/",
            num_hashes=128,
            num_buckets=14,      # 14 个 band
            hashes_per_bucket=9, # 每个 band 9 个 hash → 相似度阈值约 0.75
        ),
    ],
    tasks=128,
)

# Step 2c: 聚类
stage2c = LocalPipelineExecutor(
    pipeline=[
        MinhashDedupCluster(
            input_folder="output/stage2/buckets/",
            output_folder="output/stage2/clusters/",
        ),
    ],
    tasks=1,  # 聚类必须单进程（全局状态）
)

# Step 2d: 过滤重复
stage2d = LocalPipelineExecutor(
    pipeline=[
        MinhashDedupFilter(
            input_folder="output/stage1/clean/",
            duplicates_folder="output/stage2/clusters/",
            output_folder="output/stage2/deduped/",
        ),
    ],
    tasks=128,
)

# 按顺序执行
for stage in [stage1_executor, stage2a, stage2b, stage2c, stage2d]:
    stage.run()

NeMo-Curator：GPU 加速处理

当你有大量 GPU 资源时，NeMo-Curator 可以显著加速计算密集型步骤：

"""
NeMo-Curator：GPU 加速的去重和质量过滤
需要 NVIDIA GPU + RAPIDS 环境
"""

# pip install nemo-curator[cuda12x]

import nemo_curator as nc
from nemo_curator.datasets import DocumentDataset
from nemo_curator.modules import (
    ExactDuplicates,
    FuzzyDuplicates,
    FuzzyDuplicatesConfig,
)
from nemo_curator.filters import WordCountFilter, RepeatedParagraphsFilter
from nemo_curator.utils.distributed_utils import get_client

# 初始化 Dask-CUDA 集群（自动检测 GPU）
client = get_client(cluster_type="gpu")

# 读取数据（支持 jsonl / parquet）
dataset = DocumentDataset.read_json(
    input_path="output/stage1/clean/",
    add_filename=True,
)

# GPU 加速精确去重（基于 hash）
exact_dedup = ExactDuplicates(
    hash_method="sha256",
    id_field="doc_id",
    text_field="text",
)
dataset = exact_dedup(dataset)

# GPU 加速模糊去重（比 CPU 快 5-10 倍）
fuzzy_config = FuzzyDuplicatesConfig(
    num_hashes=128,
    num_buckets=14,
    hashes_per_bucket=9,
    jaccard_threshold=0.75,
)
fuzzy_dedup = FuzzyDuplicates(config=fuzzy_config)
dataset = fuzzy_dedup(dataset)

# 保存结果
dataset.to_json("output/deduped_gpu/")

print(f"处理完成: {len(dataset)} 文档")

NeMo-Curator 多模态扩展（2026）：NeMo-Curator 已从纯文本框架升级为支持文本、图像、视频、音频的多模态数据策展框架。NeMo Curator 26.04（2026年4月）基于 Cosmos-Xenna 0.2.0 + Ray 2.54 分布式架构，重要更新包括：

  • 视频策展：视频 embedding 生成（CLIP 或专用视频编码器）、视觉质量过滤（模糊度检测、分辨率过滤）
  • 多模态质量过滤：图像分辨率/模糊度检测、视频帧质量评估、音频 SNR 过滤
  • 跨模态去重：基于 CLIP embedding 的语义去重（解决”内容相同但像素不同”的图片重复问题）
  • text-to-image / video-language 管线：专为多模态模型训练数据设计的处理流程
  • 性能基准测试框架：综合性能基准测试；容器化部署（nvcr.io/nvidia/nemo-curator:25.09）

⚠️ 上一条的容器镜像已经没有后续版本了。 NVIDIA 自 26.09 起不再发布新的 NGC 镜像，旧镜像仍可下载但不会跟进新版本（官方原话：NVIDIA no longer publishes new NGC images starting with 26.09, though previously published images remain downloadable）。容器用户需要改为从所采用 release 的源码自行构建镜像（仓库 Dockerfile 仍在维护），音视频工作流还要在派生镜像里自行补 FFmpeg。Python 包用户注意另一处：all extra 从 26.09 起不再直接包含 cv2（可能经 vLLM 传递装上，但不能依赖），需要显式装 cv2 extra。

这反映了 2025-2026 年数据工程从纯文本走向全模态的趋势——当模型能”看图”和”听声音”之后，数据管线也必须跟上。

⚠️ 能力缺口：NeMo Curator 1.x 不提供内置 PII 脱敏。 本章前一版曾把 pii_redactor 写成 26.04 的内置 filter 类型，并配了 from nemo_curator.modules import PIIRedactor 的示例。这两条在一手文档上无法回源：截至 v1.3.0（26.07），官方”关键特性”清单覆盖文本/图像/音频/视频四类管线的加载、清洗、过滤、去重、分类与导出，PII、redact、decontaminate 三个关键词均为零命中；NGC 容器页则把 PII 处理明确写进 Known Limitations——“PII removal tools are being migrated to the new architecture”。PII 能力在 0.x 时代以 LLM NIM 教程的形式提供过（v0.9.0，2025-07），1.x 基于 Ray 重写后尚未迁回。

工程含义：PII 处理必须在自建 stage 里完成，不要指望框架内置（Presidio 一类方案与风险分级见第八章）。附带一条：1.x 的 API 形态是 Pipeline + ProcessingStage + Executor，0.x 的 Modify / ScoreFilter 调用对象已不存在，社区流传的 Modify(PIIModifier(...)) 退路在 1.x 上一样跑不通。

前面撤下的 YAML 声明式管道示例出于同一原因：官方导航与文档中检索不到 YAML 配置入口，1.x 的管道是 Python 代码态的。判断某个框架”有没有某项能力”时，以官方 Key Features 清单为准，不以教程转述为准——这一条比省下的那点配置代码更值钱。

🔄 2026-10 复核：上面的结论方向仍成立，但”零命中”的表述需要降级。 能力缺口不是二值的，它会在三个位置之间迁移。重新逐位置回源的结果：

位置截至 2026-10 的状态能否作为依赖
官方 Key Features 清单仍无 PII 条目（零命中）—
Known Limitations仍把 PII 处理标为”正在适配 Ray 后端”—
官方 tutorials/ 目录已出现 GLiNER 实现：tutorials/text/gliner-pii-redaction/gliner_pii_redactor.py，基于 nvidia/gliner-pii 模型，以自定义 ProcessingStage 形式给出，覆盖 first_name / ssn / email / 坐标等一整套实体与准标识符标签可作起点参考
官方 release notes 的 How-to 清单已列入 “LLM NIM - PII Redaction” 条目条目较简

即：教程存在 ≠ 内置能力，但”零命中”也不等于”没有任何官方路径”。工程含义不变——PII 仍然必须走自建 stage，不能指望框架内置算子；变的是”自建”这件事现在有了一份官方可回源的参考实现（一个继承 ProcessingStage 的 GLiNER stage），而不是从零设计。

这也是对上面那条方法论的边界补充：「以 Key Features 为准」这条规则防的正是把教程级方案当成内置能力，但它不该被读成”教程没有价值”。 正确的读法是分位置记录：Key Features 决定能依赖什么，Known Limitations 决定有什么坑，tutorials 决定从哪儿起步。

Roadmap：26.04 版本的发布说明中提及了后续计划中的代码策展和数学数据策展模块，将进一步扩展 NeMo Curator 的应用场景。

NeMo Curator 26.04 全栈能力矩阵（2026-04 ~ 2026-07 累积升级）

类别26.04 能力升级时间实战意义
嵌入后端vLLM 嵌入后端默认化（TextSemanticDeduplicationWorkflow），默认模型 google/embeddinggemma-300m（替代 sentence-transformers/all-MiniLM-L6-v2）2026-04推理速度提升 5-10×，可直接利用 GPU 集群空闲资源
嵌入参数新增 embedding_pretokenize、embedding_vllm_init_kwargs、model_cache_dir2026-04预 tokenize 减少重复计算；自定义 vLLM 初始化参数
推理服务InferenceServer + InferenceModelConfig 全新 API；Ray 集群内部署 OpenAI 兼容端点2026-04任意 stage 可直接 HTTP 调模型；GPU contention detection 自动强制 RayDataExecutor
生命周期Context manager 自动启动 / 清理 Ray Serve 推理服务器2026-04避免资源泄漏；与现有 pipeline 集成无侵入
Megatron 输出MegatronTokenizerWriter 一次性 tokenize + 导出 Megatron-LM 兼容 .bin/.idx（替代 preprocess_data.py）2026-04自动 dtype 选择：≤65K 词表 uint16，否则 int32
Pipeline 指标StagePerfStats.num_items_processed 文档级跟踪；pipeline.run() 返回 per-stage 文档计数2026-04阶段级性能监控（之前只能外部计时）
Workflow 接口WorkflowRunResult 标准化返回类型；WorkflowBase 抽象类统一接口2026-04所有 dedup workflow run() 返回 WorkflowRunResult 而非 None/dict
可恢复性Pipeline resumability via source-level counter checkpointing（PR #2063）2026-07-01断点续跑——3 周级别任务的关键能力
Slurm 集成Slurm array 支持（PR #2059）2026-07-01在 Slurm 集群上以 array job 形式运行 pipeline
多模态-视频Cosmos-Xenna 0.2.0：Scene detection (TransNetV2) + Fixed-stride splitting + 运动/美学过滤 + Cosmos-Embed1 嵌入2026-04视频管线完整化：H.264 编码 + 帧提取 + 运动过滤 + 美学过滤 + 语义去重
多模态-图像Aesthetic filtering + NSFW detection + CLIP 嵌入 + 语义去重2026-04图像管线完整化
多模态-音频ALM 四阶段（ALMDataBuilderStage/ALMDataOverlapStage/ALMManifestReader/ALMManifestWriterStage）；fsspec 流式 JSONL；FLEURS pipeline ~100s/4.01 tasks/s（Xenna）2026-04音频管线完整化——四阶段管线的标准化
多用户监控/tmp/nemo_curator_metrics_{uid} 默认路径；PID-file 跟踪替代进程名扫描2026-04避免多用户共享 Ray 集群冲突
可观测性start_prometheus_grafana.py 从内置 Ray dashboard factory 自动生成 4 类 dashboard2026-04Prometheus + Grafana 自动 dashboard
ImageReader 扩展ImageReaderStage RayDataExecutor 支持2026-06-13从仅 XennaExecutor 扩展到 RayDataExecutor
分类器优化Aegis / LlamaGuard classifier shared tokenizer2026-06共享基础分词器时复用 token，避免冗余 tokenize
fan-out 阶段ray_stage_spec() 让 Ray Data 在多个输出时正确重分区2026-04多输出 stage 不再出现数据倾斜

⚠️ NeMo Curator 26.07（v1.3.0）破坏性迁移速查（2026-07-27 发布）。 26.07 是自 26.02 Ray 架构重构以来的又一次大规模变更，对工程实践有直接影响的关键迁移项：

  1. Python 3.11+ 强制要求——不再支持 Python 3.10 及以下
  2. task_id 框架自管——Task.task_id 改为 init=False，用户不能再手动赋值，Curator 在每个 stage 边界自动分配并维护确定性血统
  3. EmptyTask 实例化变更——从模块级单例 EmptyTask 变为需构造的类 EmptyTask()，哨兵检测从 task is EmptyTask 改为 isinstance(task, EmptyTask)
  4. 视频预处理简化——model_does_preprocess 参数彻底移除，所有视频说明模型改用 vLLM 或 HuggingFace 原生预处理路径
  5. Ray 唯一后端——Dask 废弃，所有模态仅支持 Ray 执行器
  6. Audio SortFormer 说话人分离——新增音频说话人分离功能
  7. 推理服务器重构——InferenceModelConfig 替换为类型化的 Ray Serve/Dynamo 配置
  8. 平台支持变更——aarch64 和 macOS 不再支持 nemo_text_processing（影响 ITN/WER stage）

以上变更直接影响自定义 stage 的编写方式（task_id 自管）、部署环境要求（Python 3.11+/Ray 唯一后端）、以及流水线模板代码（EmptyTask 实例化）。升级前请务必参考官方迁移指南。

⚠️ NeMo Curator 26.09 迁移速查（官方 draft checklist）。官方在该页自陈「包版本与发布状态未确认」，引用时须沿用这一限定。与工程实践直接相关的四条：

  1. 分发方式变更——不再发布新的 NGC 镜像，容器改为从源码自建；all extra 不再含 cv2（见上文）
  2. 语义去重精度——pairwise 计算默认改为 float16、批大小 1024；若工作流需要 float32，必须同时设置 kmeans_embedding_output_dtype="float32" 与 pairwise_compute_dtype="float32"，两者混用（float16 的 KMeans 输出 + float32 的 pairwise）会在构造时直接抛 ValueError
  3. Worker 规模配置——新增 stage 级 stage.with_(num_workers_per_node=...)，用于让 worker 数跟随存活节点数；不可与 num_workers 或 Ray Data actor-pool 的 min/max/initial 同时使用
  4. 工具链与推理依赖——要求 uv 0.12 或更高；GB200 上跑 vLLM 需通过 vLLM extra 装 quack-kernels>=0.4.1

另有一条对本书方法论特别重要：默认 dtype 从 float32 变成 float16，会改变语义去重产物的数值。这正是第 30 章 §30.10「加速前应先固化 CPU 基线产物指纹」警告的场景——升级 26.09 后，先在小样本上跑一遍并把重复 ID 集合与升级前逐条对一遍，再谈加速收益。

🔬 显微镜案例：FineWeb 的基础设施规模

HuggingFace 构建 FineWeb（15T token）的基础设施规模：

指标数值
原始 CC 数据量~100TB（96 个 CC 快照）
中间文件总量~300TB
最终输出~30TB（15T token）
CPU 核心2000+ 核，分布式集群
处理时间~3 周（含调试和重试）
MinHash 签名大小~2TB
总存储消耗~500TB（包括备份和版本）
估算云成本$150K-300K

关键工程挑战：

  • 分布式 Union-Find：去重的聚类步骤需要全局状态，是整个管线的瓶颈
  • 容错：3 周的任务中，节点故障率约 5%，需要 checkpoint + 重试机制
  • 数据一致性：96 个 CC 快照之间存在 URL 重叠，跨快照去重增加了复杂度
  • 质量过滤器调参：FineWeb-Edu 的教育价值分类器经过了 >20 轮调参才达到最优

教训：数据处理的 80% 时间花在调试和重试上，而不是”正式运行”。一个好的基础设施必须有完善的日志、checkpoint 和重试机制。

28.2 大规模数据处理的工程挑战

PB 级数据处理的资源估算

处理 10PB 的预训练数据需要考虑：

存储规划：
  ├── 原始数据         ~10PB
  ├── 文本提取输出      ~3PB（HTML → 纯文本，压缩后）
  ├── 去重中间文件      ~5PB（MinHash 签名 + 桶 + 聚类）
  ├── 质量过滤中间文件   ~2PB
  ├── 最终输出          ~2PB（清洗后的训练数据）
  └── 备份 + 版本       ~5PB
  → 总计需要 ~25-30PB 可用存储

计算资源：
  ├── 文本提取          ~1000 CPU·天（IO 密集）
  ├── 语言识别          ~200 CPU·天（fasttext 推理）
  ├── 质量过滤（规则）   ~100 CPU·天
  ├── 质量过滤（模型）   ~50 GPU·天（如 FineWeb-Edu 分类器）
  ├── MinHash 签名      ~500 CPU·天 + 2TB 内存
  ├── LSH 分桶          ~100 CPU·天
  ├── 聚类（Union-Find） ~50 CPU·天 + 4TB 内存（瓶颈！）
  └── 最终过滤和输出     ~100 CPU·天
  → 总计 ~2000 CPU·天 + 50 GPU·天

成本估算（AWS 参考价）：
  ├── 存储 (S3)：30PB × $0.023/GB/月 ≈ $690K/月
  ├── 计算 (EC2)：
  │   ├── CPU: 2000 天 × ~$2/小时 × 24h ≈ $96K
  │   └── GPU: 50 天 × ~$30/小时 × 24h ≈ $36K
  ├── 网络传输：~$20K
  └── → 一次完整管线运行 ≈ $150K-250K（不含存储月费）

使用 1000 核 CPU 集群的并行化：
  → 2000 CPU·天 / 1000 核 ≈ 2 天计算时间
  → 加上 IO 和调度开销 ≈ 3-5 天端到端

上面这套估算的执行后端是 CPU 集群（FineWeb 口径：2000+ 核、约 3 周）。同一件事走 GPU 加速路径是另一个量级——NVIDIA 官方（NeMo Curator v1.3.0 / 26.07）给出的数字是 32 张 H100、0.5 小时去重 1.96 万亿 token（RedPajama V2 规模），以及 30 个 CPU 节点、不到 38 小时把 Common Crawl 从 2.8TB 处理到 0.52TB，官方口径称数据削减最高约 80%。

两组数字不能互相换算，引用时必须标明执行后端：

  • 官方的 1.96T / 0.5h 是纯去重环节的口径，不含文本提取、质量过滤与写出，端到端耗时远高于此；
  • FineWeb 的 3 周是端到端管线口径，还包含官方口径里不体现的调试与重试（FineWeb 团队自己的说法是”80% 的时间花在调试和重试上”）；
  • 削减率也不同源：FineWeb 是 100TB → 30TB，官方 CC 口径是 2.8TB → 0.52TB，两者的输入快照范围与过滤强度都不一致。

做预算时把这两条当作参考区间的两端，别拿 GPU 侧的吞吐去乘 CPU 侧的工期。

还有一层前提要写在估算旁边：上面三档（FineWeb 的 CPU 端到端、NVIDIA 的 GPU 纯去重、本节的自有集群推演）都是自有或长期租用集群的口径——成本模型的核心是”集群利用率”，先把机器拿在手里，再想办法把它填满。2026 年下半年出现的 serverless 编排（datatrove JobsPipelineExecutor，见 28.1 决策树最上面那条分支）把固定成本换成了按次计费，成本模型随之变成”单价 × 分片数”。对月级以下的短管线，这个替换往往是决定性的；对持续跑数月的管线，自有集群仍然更划算。本节的估算方法本身仍然有效，但引用时须标明它属于哪一种成本模型。

⚠️ 待验证：本段为推断性补充，本章尚未做两种成本模型的逐项对照（需要同一管线的”集群摊销价”与”serverless 单价 × 分片数”两套数字才能定拐点）。

分布式去重的核心实现

去重是数据管线中计算量最大且工程最复杂的步骤：

"""
分布式 MinHash 去重的核心组件
生产环境中通常用 datatrove/dolma 的实现，
这里展示关键算法以便理解原理
"""
import hashlib
import struct
from typing import Optional

class DistributedMinHashDedup:
    """分布式 MinHash + LSH 去重"""
    
    def __init__(
        self,
        num_hashes: int = 128,
        num_bands: int = 14,
        rows_per_band: int = 9,
        ngram_size: int = 5,
        seed: int = 42,
    ):
        """
        参数选择指南：
        - num_hashes: 越大越精确，但计算和存储成本更高
        - num_bands × rows_per_band = num_hashes
        - 相似度阈值 ≈ (1/num_bands)^(1/rows_per_band)
        - 14 bands × 9 rows → 阈值 ≈ 0.75
        - 20 bands × 5 rows → 阈值 ≈ 0.55（更激进的去重）
        """
        assert num_bands * rows_per_band <= num_hashes
        self.num_hashes = num_hashes
        self.num_bands = num_bands
        self.rows_per_band = rows_per_band
        self.ngram_size = ngram_size
        self.seed = seed
        
        # 生成随机哈希参数（a*x + b mod p）
        import random
        rng = random.Random(seed)
        self.hash_params = [
            (rng.randint(1, 2**31 - 1), rng.randint(0, 2**31 - 1))
            for _ in range(num_hashes)
        ]
    
    def compute_signature(self, text: str) -> list[int]:
        """
        Step 1: 计算文档的 MinHash 签名
        - 提取 word-level n-grams（对中文需要先分词）
        - 对每个 n-gram 计算哈希
        - 对每个哈希函数取最小值
        """
        # 提取 shingles（n-grams）
        words = text.split()
        shingles = set()
        for i in range(len(words) - self.ngram_size + 1):
            shingle = " ".join(words[i:i + self.ngram_size])
            shingle_hash = struct.unpack(
                '<I', hashlib.md5(shingle.encode()).digest()[:4]
            )[0]
            shingles.add(shingle_hash)
        
        if not shingles:
            return [0] * self.num_hashes
        
        # 计算 MinHash 签名
        LARGE_PRIME = 2**31 - 1
        signature = []
        for a, b in self.hash_params:
            min_hash = min((a * s + b) % LARGE_PRIME for s in shingles)
            signature.append(min_hash)
        
        return signature
    
    def get_lsh_buckets(self, signature: list[int]) -> list[str]:
        """
        Step 2: LSH 分桶
        将签名分成 num_bands 个 band，
        每个 band 哈希到一个桶中
        """
        buckets = []
        for band_idx in range(self.num_bands):
            start = band_idx * self.rows_per_band
            end = start + self.rows_per_band
            band = tuple(signature[start:end])
            bucket_key = f"band_{band_idx}_{hashlib.md5(str(band).encode()).hexdigest()}"
            buckets.append(bucket_key)
        return buckets


class DistributedUnionFind:
    """
    分布式 Union-Find（并查集）
    用于将候选重复对合并成连通分量
    
    在生产环境中，这一步通常是瓶颈，
    因为需要全局状态（不能简单并行化）
    
    解决方案：
    1. 分批处理 + 迭代收敛
    2. 使用 Redis/Memcached 作为分布式存储
    3. 使用图数据库（如 Neo4j）处理大规模连通分量
    """
    
    def __init__(self):
        self.parent = {}
        self.rank = {}
    
    def find(self, x: str) -> str:
        """查找根节点（带路径压缩）"""
        if x not in self.parent:
            self.parent[x] = x
            self.rank[x] = 0
            return x
        if self.parent[x] != x:
            self.parent[x] = self.find(self.parent[x])
        return self.parent[x]
    
    def union(self, x: str, y: str):
        """合并两个集合（按秩合并）"""
        rx, ry = self.find(x), self.find(y)
        if rx == ry:
            return
        if self.rank[rx] < self.rank[ry]:
            rx, ry = ry, rx
        self.parent[ry] = rx
        if self.rank[rx] == self.rank[ry]:
            self.rank[rx] += 1
    
    def get_clusters(self) -> dict[str, list[str]]:
        """获取所有连通分量"""
        from collections import defaultdict
        clusters = defaultdict(list)
        for node in self.parent:
            root = self.find(node)
            clusters[root].append(node)
        return dict(clusters)

28.3 数据管理与版本控制

为什么数据版本管理至关重要

没有版本管理的后果（真实案例）：

场景：团队花了 2 个月训练了一个模型，效果很好。
     3 个月后想在新硬件上复现，发现……

  ❌ "这个模型的训练数据是哪个版本的？"
     → 没人记得。数据目录被覆盖过多次。

  ❌ "质量过滤用的是什么参数？"
     → 当时的脚本被修改了，原始参数没有记录。

  ❌ "去重的阈值是 0.7 还是 0.8？"
     → 不确定，两个版本的脚本都在。

  ❌ 结果：花了 3 周重新处理数据，但新模型的效果和原来差了 2 个点。
     原因无法排查，因为没有完整的数据血缘。

工具选择与实践

按规模选择版本管理工具：

小规模（< 1TB）：DVC（Data Version Control）
  ├── 与 Git 无缝集成
  ├── 文件级版本管理
  ├── 支持 S3/GCS/Azure 远程存储
  └── 最简单的起步方案

中规模（1-100TB）：lakeFS
  ├── Git-like 分支模型（branch/commit/merge）
  ├── 直接操作 S3/GCS 对象存储
  ├── 零拷贝分支（只记录差异）
  ├── 支持 ACID 事务
  └── 可以在"数据分支"上做实验而不影响主干

大规模（> 100TB）：Delta Lake / Apache Iceberg
  ├── 与 Spark 生态深度集成
  ├── PB 级数据的高效版本管理
  ├── 支持 Time Travel（回溯到任意历史版本）
  ├── Schema Evolution（字段变更不破坏兼容性）
  └── 已被 Databricks/Netflix/Apple 大规模使用

最小可行方案：Manifest 文件
  ├── 如果上面的工具都太重
  ├── 至少为每个数据版本维护一个 manifest.json
  ├── 记录：文件列表、sha256、处理参数、生成时间
  └── 这是版本管理的"地板"，不能再低了

2026-05 选型决策矩阵：DVC / lakeFS / Delta Lake 定位差异显著，选型需考虑”团队栈 + 规模 + 一致性需求”三个维度：

工具定位适用规模Git 语义Spark 集成ACID 事务学习成本
DVC轻量实验管理GB-TB强（基于 Git）弱弱低（Git 用户零成本）
lakeFSPB 级数据湖 Git 语义TB-PB强（独立元数据层）中中中（需理解 Git 语义）
Delta LakeSpark 生态表格式TB-PB弱强强（optimistic concurrency）中（需 Spark 基础）
Apache Iceberg跨引擎表格式TB-PB弱强强中（社区生态增长中）
Apache Hudi实时数据湖TB-PB弱强强高（Copy-on-Write/Merge-on-Read 复杂度）

选型决策树（2026-05 升级版）：

  • 轻量实验 / 小团队 / 单机可管理 → DVC（Git 语义 + 文件级版本）
  • PB 级数据湖 + 跨团队协作 + Git 式分支/合并/回滚 → lakeFS（独立元数据层，零拷贝分支）
  • 已有 Spark 生态 + ACID 需求 + Time Travel → Delta Lake
  • 跨引擎（Spark/Trino/Flink） + 演进 schema → Apache Iceberg
  • 实时 upsert + CDC 场景 → Apache Hudi
  • 资源极度受限 → Manifest 文件（最低门槛）
"""
最小可行数据版本管理：Manifest 文件
当你没有条件用 DVC/lakeFS 时，至少做到这一步
"""
import json
import hashlib
import os
from datetime import datetime

def create_data_manifest(
    data_dir: str,
    pipeline_config: dict,
    description: str = "",
) -> dict:
    """
    创建数据版本清单
    
    Args:
        data_dir: 数据目录
        pipeline_config: 处理管线的完整配置
        description: 人类可读的版本描述
    """
    manifest = {
        "version": datetime.now().strftime("%Y%m%d_%H%M%S"),
        "description": description,
        "created_at": datetime.now().isoformat(),
        "data_dir": data_dir,
        "pipeline_config": pipeline_config,
        "files": [],
        "stats": {"total_files": 0, "total_bytes": 0},
    }
    
    for root, dirs, files in os.walk(data_dir):
        for f in sorted(files):
            filepath = os.path.join(root, f)
            file_size = os.path.getsize(filepath)
            
            # 对大文件只计算前 1MB 的 hash（节省时间）
            with open(filepath, 'rb') as fh:
                chunk = fh.read(1024 * 1024)
                file_hash = hashlib.sha256(chunk).hexdigest()
            
            manifest["files"].append({
                "path": os.path.relpath(filepath, data_dir),
                "size_bytes": file_size,
                "sha256_1mb": file_hash,
            })
            manifest["stats"]["total_files"] += 1
            manifest["stats"]["total_bytes"] += file_size
    
    manifest["stats"]["total_size_human"] = _human_readable_size(
        manifest["stats"]["total_bytes"]
    )
    
    # 保存 manifest
    manifest_path = os.path.join(data_dir, f"manifest_{manifest['version']}.json")
    with open(manifest_path, 'w') as f:
        json.dump(manifest, f, indent=2, ensure_ascii=False)
    
    return manifest

def _human_readable_size(size_bytes: int) -> str:
    for unit in ['B', 'KB', 'MB', 'GB', 'TB', 'PB']:
        if abs(size_bytes) < 1024:
            return f"{size_bytes:.1f} {unit}"
        size_bytes /= 1024
    return f"{size_bytes:.1f} EB"

# 使用示例
manifest = create_data_manifest(
    data_dir="/data/pretrain_v3/",
    pipeline_config={
        "source": "CC-MAIN-2024-18 to CC-MAIN-2024-30",
        "extractor": "trafilatura 1.8",
        "language_filter": {"langs": ["zh", "en"], "threshold": 0.65},
        "quality_filter": "gopher + fineweb-edu (threshold=3)",
        "dedup": "minhash-128-14-9 (jaccard=0.75)",
        "total_tokens_before_dedup": "5.2T",
        "total_tokens_after_dedup": "3.1T",
    },
    description="预训练数据 v3：2024年CC数据，新增 FineWeb-Edu 过滤",
)

数据血缘追踪

数据血缘（Data Lineage）的完整设计：

每条数据的血缘记录应包含：

{
  "doc_id": "cc_2024_18_seg5_doc_3847291",
  "lineage": [
    {
      "stage": "extraction",
      "timestamp": "2025-01-15T10:32:00Z",
      "tool": "trafilatura 1.8.0",
      "input": "warc://CC-MAIN-2024-18/segments/5/warc/00042.warc.gz#offset=38472",
      "params": {"include_comments": false, "include_tables": true},
      "output_size": 4523,
    },
    {
      "stage": "language_filter",
      "timestamp": "2025-01-15T11:05:00Z",
      "tool": "fasttext lid.176.bin",
      "result": {"lang": "zh", "score": 0.94},
      "decision": "keep",
    },
    {
      "stage": "quality_filter",
      "timestamp": "2025-01-15T12:30:00Z",
      "tool": "gopher_quality_v2",
      "scores": {"word_count": 1234, "avg_word_len": 4.2, "stop_ratio": 0.12},
      "decision": "keep",
    },
    {
      "stage": "dedup",
      "timestamp": "2025-01-16T08:00:00Z",
      "tool": "minhash_dedup",
      "params": {"threshold": 0.75, "num_hashes": 128},
      "cluster_id": "cluster_89247",
      "cluster_size": 3,
      "decision": "keep (representative)",
    },
  ]
}

→ 有了血缘记录，任何异常都可以追溯到根源

28.4 自动化数据管线（CI/CD for Data）

管线编排

"""
数据管线编排：使用 DAG（有向无环图）管理阶段依赖
轻量级方案，不依赖 Airflow 等重框架
"""
import subprocess
import time
import json
from pathlib import Path
from dataclasses import dataclass, field

@dataclass
class PipelineStage:
    name: str
    command: str
    depends_on: list[str] = field(default_factory=list)
    timeout_hours: float = 24.0
    retries: int = 3
    checkpoint_dir: str = ""

class DataPipelineOrchestrator:
    """轻量级数据管线编排器"""
    
    def __init__(self, stages: list[PipelineStage], output_dir: str):
        self.stages = {s.name: s for s in stages}
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.status = {}  # name → "pending" | "running" | "success" | "failed"
    
    def run(self):
        """按依赖顺序执行所有阶段"""
        for name in self._topological_sort():
            stage = self.stages[name]
            
            # 检查是否已有 checkpoint
            if self._has_checkpoint(name):
                print(f"[SKIP] {name}: checkpoint exists")
                self.status[name] = "success"
                continue
            
            # 检查依赖是否都成功
            for dep in stage.depends_on:
                if self.status.get(dep) != "success":
                    print(f"[SKIP] {name}: dependency {dep} not ready")
                    self.status[name] = "skipped"
                    continue
            
            # 执行
            print(f"[START] {name}")
            self.status[name] = "running"
            success = self._execute_with_retry(stage)
            self.status[name] = "success" if success else "failed"
            
            if not success:
                print(f"[FAIL] {name}: all retries exhausted")
                break
        
        self._save_report()
    
    def _execute_with_retry(self, stage: PipelineStage) -> bool:
        for attempt in range(stage.retries):
            try:
                start = time.time()
                result = subprocess.run(
                    stage.command, shell=True,
                    timeout=stage.timeout_hours * 3600,
                    capture_output=True, text=True,
                )
                elapsed = time.time() - start
                
                if result.returncode == 0:
                    print(f"  [{stage.name}] success in {elapsed/60:.1f} min")
                    self._save_checkpoint(stage.name, elapsed)
                    return True
                else:
                    print(f"  [{stage.name}] attempt {attempt+1} failed: {result.stderr[:200]}")
            except subprocess.TimeoutExpired:
                print(f"  [{stage.name}] attempt {attempt+1} timeout")
        return False
    
    def _topological_sort(self) -> list[str]:
        """拓扑排序"""
        visited = set()
        order = []
        def visit(name):
            if name in visited: return
            visited.add(name)
            for dep in self.stages[name].depends_on:
                visit(dep)
            order.append(name)
        for name in self.stages:
            visit(name)
        return order
    
    def _has_checkpoint(self, name: str) -> bool:
        return (self.output_dir / f"{name}.checkpoint").exists()
    
    def _save_checkpoint(self, name: str, elapsed: float):
        ckpt = {"stage": name, "elapsed_sec": elapsed, "timestamp": time.time()}
        (self.output_dir / f"{name}.checkpoint").write_text(json.dumps(ckpt))
    
    def _save_report(self):
        report = {"stages": self.status, "timestamp": time.time()}
        (self.output_dir / "pipeline_report.json").write_text(
            json.dumps(report, indent=2)
        )
        print(f"\n管线报告已保存: {self.output_dir / 'pipeline_report.json'}")

# 使用示例
stages = [
    PipelineStage("extract", "python extract.py --input /data/warc/ --output /data/text/"),
    PipelineStage("lang_filter", "python filter_lang.py", depends_on=["extract"]),
    PipelineStage("quality_filter", "python filter_quality.py", depends_on=["lang_filter"]),
    PipelineStage("dedup_sign", "python dedup_minhash.py --stage sign", depends_on=["quality_filter"]),
    PipelineStage("dedup_bucket", "python dedup_minhash.py --stage bucket", depends_on=["dedup_sign"]),
    PipelineStage("dedup_cluster", "python dedup_minhash.py --stage cluster", depends_on=["dedup_bucket"]),
    PipelineStage("dedup_filter", "python dedup_minhash.py --stage filter", depends_on=["dedup_cluster"]),
    PipelineStage("stats", "python compute_stats.py", depends_on=["dedup_filter"]),
]
orchestrator = DataPipelineOrchestrator(stages, "/data/pipeline_state/")
# orchestrator.run()

质量门控（Quality Gates）

每个阶段之间设置质量门控，自动检测异常：

文本提取后：
  ├── 检查：平均文档长度是否在合理范围（500-5000 词）
  ├── 检查：空文档比例 < 5%
  └── 检查：非 UTF-8 文档比例 < 0.1%

语言过滤后：
  ├── 检查：目标语言比例是否符合预期（如中英混合 60:40）
  ├── 检查：被过滤掉的比例是否异常（正常 30-50%）
  └── 检查：低置信度样本的语言分布

去重后：
  ├── 检查：去重率是否在预期范围（正常 20-40%）
  ├── 检查：最大连通分量大小是否异常（> 10000 说明阈值太松）
  └── 检查：去重前后的主题分布是否一致

质量过滤后：
  ├── 检查：过滤率是否在预期范围
  ├── 检查：各质量分数的分布是否合理
  └── 检查：过滤后数据的随机样本人工审核（50 条）

⚠️ 训练基础设施的可靠性验证：Poolside 的”激进不信任”实践（2026-07 公开）。数据工程不仅要保证数据管线的正确性，还需要与训练基础设施的验证机制协同。Poolside 团队在 Laguna S（118B 总参 / 8B 活跃 / 4000 GPU × 30T token）训练中实践了一套”激进不信任”验证方法：在每次 DDP 同步后对所有副本的模型权重计算哈希——一旦哈希不一致，立即崩溃训练以防止静默损坏扩散。这套机制发现了三类典型故障：(1) 物理损坏的 GPU 导致静默数据损坏（两个完全相同的训练产生不同损失曲线）；(2) BF16 累加精度不足导致损失停滞（修复：单次累加改用 FP32）；(3) FP8 内核竞态条件导致 0.5% 梯度每步静默损坏（哈希校验无法检测此类故障，因为各副本处理不同数据）。这类基础设施层面的问题可能在浪费数百万美元计算资源后才被发现——数据工程师需要理解这些边界条件，才能正确解读训练结果、区分”数据问题”与”基础设施问题”。

Agent 时代的管线产物可治理性：NL2Pipeline gap（2026-07）。28.4 讲的 CI/CD for Data 解决的是「人写的管线如何版本化与门控」。但当管线由 coding agent 生成时，会出现一个本节尚未覆盖的问题：agent 产出的是一次性脚本——能跑、能完成当前任务，但无法在平台里审计、编辑、复用，也变不成标准数据资产。北大 OpenDCAI 把这个脱节命名为 NL2Pipeline gap（arXiv:2607.16617，2026-07-18）。

DataFlow-Harness 的解法是让 agent 不生成脚本，而提交类型化的增量变更（增删算子、改参数、连边）；每次变更须过「DAG 保持无环 + 相邻算子 schema 兼容」校验才落盘。同时用 MCP 层把实时算子注册表与当前管线状态暴露给 agent，用 WebSocket 让对话式编辑与可视化 DAG 编辑器共享同一状态。

实测（12 个数据工程任务 × 6 类工业场景 × 10 次 = 120 runs，底座固定 Claude Opus 4.7）：端到端通过率 93.3%；相比让 agent 自由写脚本，成本降低 72.5%、延迟降低 49.9%，通过率仅差 0.9 个百分点。

这条结论反直觉但值得记住：约束 agent 在平台语义内工作，比让它自由写代码更便宜——因为 agent 不再烧 token 重复推导可以直接查询的结构，也不再重试那些根本不可能通过类型检查的管线。消融显示，注入程序性知识的 Skills 层贡献了最大单块增益（+4 个百分点），且只在需要隐式过程性知识的任务上显现。

⚠️ 论文自陈的边界：校验只保证结构有效性，不保证语义正确性、端点可用性或输出质量。把它接进生产时，schema 兼容 ≠ 数据质量达标，下游校验仍不可省。

28.5 Agent 数据工程：新兴方向

传统数据工程的处理对象是”文档”——从网页提取文本，清洗、去重、过滤。但随着 LLM Agent 成为主流应用形态，一种新的数据类型正在涌现：Agent 轨迹数据。

2026 年 2 月的研究（On Data Engineering for Scaling LLM Terminal Capabilities）指出，Agent/终端能力的训练数据与传统预训练数据有根本差异：

维度传统文本数据Agent 轨迹数据
数据单元文档/段落多步交互序列（思考→行动→观察→思考→…）
包含内容自然语言文本工具调用、API 响应、环境反馈、错误信息
质量信号文本流畅性、信息密度任务是否成功完成、步骤是否高效
去重挑战文本相似度相同任务的不同解法应保留（多样性 > 去重）
Scaling更多数据 → 更好更多成功轨迹的多样性 → 更好

目前现有框架（datatrove、NeMo-Curator、dolma）都是围绕”文档处理”设计的，Agent 轨迹的处理需要不同的管线架构——需要解析结构化的交互序列、评估每一步的合理性、以及处理工具调用的版本兼容性问题。这是一个工程上尚未成熟但需求快速增长的方向。

MCP 协议与 Agent 数据采集：MCP（Model Context Protocol）已成为 AI Agent 与外部工具/数据源交互的行业标准协议。MCP 工具调用记录天然具有结构化格式（工具名→参数→结果），是理想的 Agent 轨迹数据来源。通过 MCP Session 的完整消息历史，可以高效采集高质量的 Agent 交互数据。建议将 MCP 协议数据采集纳入 Agent 数据工程的标准管线——当 Agent 的工具调用有了统一的协议标准，数据工程就可以在此基础上构建标准化的采集、清洗和质量评估流程。

生态规模（2026-10 口径，务必带量纲引用）

指标数值时点量纲
SDK 下载约 9,700 万 / 月2026-10月下载量，约为半年前的 3 倍
活跃公开 server约 10,0002026-10公开可得的 server 数
npm + GitHub 上的 server13,000+2026-05含未进注册中心的包
官方注册中心记录9,6522026-10注册中心数 < 生态真实规模（私有企业 server 不出现）

制度性节点两个：2025-12 MCP 被捐给 Linux Foundation（现由 Agentic AI Foundation 托管）；2026-03 新增无状态 HTTP transport。

⚠️ 上表数字主要来自单一非正式普查（且该普查与一个 registry 产品的发布同文），方向与量级可信，精确数字不宜当作审计级数据引用。更值得留意的是量纲：旧表述「9700 万安装量」与新口径「月下载 9700 万」数值高度接近但含义完全不同，混用会让读者把一个月度流量读成累计存量。

新增一层：采集侧的信任问题。 上面建议”把 MCP 数据采集纳入标准管线”，但工具池从几百涨到一万之后，约束已经换位——不再是”有没有工具可采”，而是”敢不敢把凭证交给这个 server”。公开调查显示有相当比例的面向公网的 MCP server 完全没有认证（不同调查给出的比例从约四分之一到约四成不等，样本与口径不同，精确值待核），同时存在 tool poisoning：把指令藏进工具描述或返回内容，让模型把它当作任务的一部分执行。对数据工程管线的 direct 后果是三条：

  1. server 准入——维护状态、认证方式、最小权限三项在接入前逐项登记，而不是接进来再说；
  2. 工具描述按不可信输入处理——工具名、参数说明、返回值都可能是注入载体，采集时保留原文但不得直接拼进下游 prompt；
  3. 逐次调用审计留痕——单次调用级的日志，而不是只记会话级结果。

这三条与第十三章的溯源字段是同一套体系：Agent 轨迹数据一旦进入训练集，“这条轨迹来自哪个 server、当时它是谁在维护”就必须可回溯，否则污染无法定位。另外，生态侧自己也把发现（discovery）列为主要瓶颈——从”需要一个 Postgres 连接器”到”找到一个可信的包”之间没有可靠路径，这层目前只能靠团队自建白名单 registry 来补。

Agent 数据工程三大 2026 范例

1. TOUCAN — 1.5M 工具智能体轨迹数据集（AGI Frontier, 2026-07）

  • 规模：1.5M 工具智能体轨迹（150 万条），覆盖近 500 个真实世界 MCP 环境
  • 场景：并行/多步工具调用、多轮对话、边缘案例——目前最大公开可用工具智能体数据集
  • 方法：从近 500 个真实 MCP 环境合成 1.5M 轨迹，无需人类标注
  • 结果：在 BFCL v3 基准上训练的开源模型超越闭源模型（SOTA）
  • 数据工程意义：是”工具使用合成”第三代范式（stateless SFT → interactive trajectory → real-MCP environment）的代表

2. ASTRA — 全自动 Agent 训练框架（链家科技, arXiv:2601.21558, 2026-01-30）

  • 规模：1,585 个 MCP 服务器 / 19,036 个工具 / 41 个领域
  • 方法：
    • SFT 阶段：工具调用图合成多轮结构化轨迹（树状展开而非线性拼接）
    • RL 阶段：QA 分解构造可验证 Python 环境（避免 hallucinated 反馈）
    • 统一训练：SFT + 在线 RL 两阶段统一训练框架
  • 结果：14B/32B 模型在 BFCL v3 / τ²-Bench / ACEBench 达 SOTA，接近闭源模型水平
  • 开源：完整 pipelines + environments + trained models at https://github.com/LianjiaTech/astra
  • 数据工程意义：“端到端 agent 数据合成”代表——MCP 工具池构建 → 工具调用图轨迹合成 → 可验证 RL 环境构造三阶段方法论

3. Nemotron-CC — 商业级 CC 过滤范式（NVIDIA, 2026）

  • 规模：6.3T token
  • 4 阶段蒸馏：领域分类 → 质量分类器集成 → 去重 → 合成数据
  • 结果：80 亿参数模型在 MMLU 提升 5 分、ARC-Challenge 提升 3.1 分；比 DCLM 在 MMLU 高 5.6 分
  • ASI-Evolve（上海交大 GAIR-NLP, 2026-03）：通过自主清洗使 MMLU 提升 18.64 个百分点（504B token Nemotron-CCASI+ vs 168B 低质原数据），证明”数据质量 > 数据数量”
  • 数据工程意义：“商业级 CC 过滤”代表——从 6.3T 原始数据蒸馏出 504B 高质量数据的工艺链

这三个范例分别覆盖”工具 agent 数据 / 端到端 agent 训练 / 商业级 CC 蒸馏”三个方向，构成了 2026 年 Agent 数据工程的三大支柱。

🔬 合成数据编排参考模式：Poolside Hive 六组件架构（2026-07 公开）。TOUCAN / ASTRA / Nemotron-CC 分别代表”端到端环境采集 / 任务树合成 / 多阶段蒸馏”三条路线，而 Poolside 的 Hive 框架将这些路线的共性抽象为六个可组合组件：Seeds（原始输入）→ Metadata（任务类型/难度等副指令）→ Generator（LLM Agent 加工具）→ Filters/Validators（正确性/去重/格式）→ Orchestrator（动态调度——决定下一步执行哪个 Agent、是否注入新元数据、质量门控循环）→ Supervisor（全局终止条件与部分结果聚合）。四种流水线形态对应不同成本-质量权衡：Rephrasing（单 Agent 低成本）、Multi-stage Aggregation（分解复杂任务为子步骤）、Cross-domain Porting（模态转换如数学→代码）、Multi-turn Evolution（Evolver+Judge 迭代）。对数据工程团队的启示：Orchestrator 层的动态调度将合成数据生成从”固定多阶段脚本”升级为”自适应质量驱动的智能工作流”，尤其适合 Agent 轨迹这类需要”分解-验证-回退”闭环的复杂数据。（Hive 所属的”下一代数据工程平台”构想，本书在讨论平台化趋势时会再展开）

28.5.1 2026 数据工程工具栈全景图

原始数据层:
  └── Common Crawl WARC / Web / GitHub / arXiv / 书籍 / 中文站点
        ↓
分布式存储: S3 / GCS / HDFS / MinIO + lakeFS / Delta Lake / Iceberg 版本控制
        ↓
处理框架:
  ├── datatrove 0.10.x (轻量学术研究；执行器四选一：Local / Slurm / Ray / Jobs(serverless))
  ├── dolma (Allen AI 学术 + 工业混合)
  └── NeMo Curator 26.09 (NVIDIA 工业级：vLLM 嵌入 + Ray Serve + 多模态；自 26.09 起无新 NGC 镜像，需从源码自建容器)
        ↓
质量过滤:
  ├── 规则式: fastText 语言识别 / KenLM 困惑度 / URL 黑名单
  └── 模型式: NeMo classifiers / Qwen 质量分类器 / FineWeb-Edu classifier
        ↓
去重:
  ├── Exact: SHA256 (文本精确重复)
  ├── MinHash LSH (近似文本)
  └── Semantic (GPU 加速: vLLM + embeddinggemma-300m)
        ↓
合成数据:
  ├── NeMo Data Designer v0.5.0 (声明式合成)
  ├── MCP 工具池构建 (MCP-Flow / TOUCAN / ASTRA)
  ├── DataFlow (北大 OpenDCAI: 近 200 算子 / 六条领域管线，LLM 在环的合成+精炼，见 28.1)
  └── 持续学习: RODS 奖励驱动在线合成 (arXiv:2606.19047)
        ↓
输出格式:
  ├── Megatron .bin/.idx (NeMo Curator 26.04 MegatronTokenizerWriter)
  ├── Arrow / Parquet
  └── JSONL (调试 / 评估)
        ↓
训练: 分布式训练 (Megatron / NeMo / DeepSpeed)

工具栈选型原则（2026-07）：

  1. 学术研究、小数据量实验 → datatrove + DVC + JSONL
  2. 开源数据集二次过滤 → dolma + DVC/lakeFS + MinHash
  3. 工业级生产 → NeMo Curator 26.09 + Delta Lake/Iceberg + vLLM 嵌入
  4. Agent 数据合成 → MCP 工具池 + ASTRA / TOUCAN / RODS
  5. 无自有集群的短管线 → datatrove + JobsPipelineExecutor（按次计费，见 28.1 决策树）

28.5.2 2026 开放预训练数据集全景

数据工程基础设施的能力最终需要回答”用什么数据训练”。2025-2026 年间，开放预训练数据集的规模发生了4-5× 扩张——这一规模量级是任何数据工程管线必须响应的基础事实。

数据集规模维护者License关键特性
FineWeb 2~10T tokensHugging FaceODC-BY多语言高质量过滤（FineWeb 1.5T → 2 ~10T，6.7× 扩张）
Common Pile~1T tokens (8TB)Allen AI + EleutherAI + 社区Permissive (curated)明确版权清理（fully-permissive），训练 Comma v0.1-1T/2T；OLMo 2 另用 Dolma 3
Dolma 3~6T tokensAllen AIODC-BYOLMo 2 训练用（Dolma 1 1.6T → 3 ~6T，3.75× 扩张）
RedPajama 2~30T tokensTogether AIVarious全谱系（v1 1.2T → 2 ~30T，25× 扩张）
Nemotron-CC~6.3T tokensNVIDIAMulti-licence4 阶段蒸馏（领域分类→质量分类器→去重→合成），商业级范式
DCLM (DataComp-LM)~3.8T tokensCommunity + Apple + AI2CC-BY-4.0学术级质量过滤 baseline
Zyda 2~5T tokensZyphraPermissive紧凑高质量
SlimPajama~627B tokensCerebrasApache 2.0RedPajama 2 严格去重 + 过滤版

关键事实：

  • 2025-2026 数据集规模 4-5× 扩张：FineWeb 1.5T → 2 ~10T、RedPajama 1.2T → 2 ~30T、Dolma 1 1.6T → 3 ~6T
  • 前沿模型实际训练 token 量（与数据集规模脱钩）：
模型训练 token相对 Chinchilla 倍数
Llama 3.1 8B15.6T~1950×（7B 模型 1875×）
DeepSeek-V314.8T~370×（671B 模型）
Qwen 3~36T~300×（最大变体）
GPT-5 / Claude 5未公开推测 30-50T 级别
  • 趋势：从”原始 CC dump → 重过滤+去重+分类器评分子集”——> “商业级 CC 蒸馏” + “fully-permissive 合规数据集” + “多语言高质量过滤”三路并行

对数据工程的指导：

  1. 中小团队：直接使用 FineWeb 2 / DCLM-baseline / SlimPajama 作为底座，不从 Common Crawl WARC 原始数据开始
  2. 中文场景：WanjuanCC（上海 AI Lab / OpenDataLab，~1T tokens 高质量 / 100B tokens 开源，Apache-2.0，arXiv:2402.19282）+ CCI3-Data + ChineseWebText + SkyPile-150B——从”自建爬虫+清洗”演进到”高质量开源+选择性二次过滤”
  3. 多模态：The Stack v2 ~3T tokens（代码）+ Cosmos 多模态管线（图像/视频/音频）
  4. Agent 数据：TOUCAN（1.5M 工具轨迹）+ ASTRA（19K 工具 SFT+RL）+ MCP-Flow（11K+ tools / 68K+ pairs）

28.6 监控与告警

数据管线仪表盘

一个生产级数据管线需要实时监控：

关键监控指标：

处理进度：
  ├── 已处理 / 总任务数
  ├── 当前处理速度（文档/秒、GB/小时）
  ├── 预计完成时间
  └── 各阶段的进度条

数据质量：
  ├── 各阶段的通过率趋势图
  ├── 异常文档的实时采样展示
  ├── 语言分布饼图
  └── 文档长度分布直方图

资源使用：
  ├── CPU / GPU 利用率
  ├── 内存使用（特别关注去重阶段）
  ├── 磁盘 IO
  └── 网络带宽（如果涉及远程存储）

告警规则：
  ├── 🔴 任何阶段失败 → 立即通知
  ├── 🟡 处理速度下降 > 50% → 调查原因
  ├── 🟡 某阶段过滤率异常偏高/偏低 → 检查过滤器
  └── 🟢 每日自动生成处理报告

动手环节：搭建一个迷你数据处理管线

目标：用 datatrove 风格的模块化设计，搭建一个处理小批量文本数据的管线。

"""
动手练习：迷你数据处理管线
处理 100 个模拟网页文本，体验完整的管线流程
"""
import json
import re
import hashlib
from collections import Counter

# ====== 模拟数据 ======
raw_documents = [
    {"id": f"doc_{i:04d}", "url": f"https://example.com/page{i}", 
     "text": f"这是第 {i} 篇文档的内容。" * (i % 20 + 5)}
    for i in range(100)
]
# 故意加入一些"脏数据"
raw_documents[10]["text"] = ""  # 空文档
raw_documents[20]["text"] = raw_documents[21]["text"]  # 完全重复
raw_documents[30]["text"] = "buy now! click here! 💰💰💰" * 50  # 垃圾内容
raw_documents[40]["text"] = "a" * 3  # 过短

# ====== 管线 Step 1：基础过滤 ======
def basic_filter(docs: list[dict], min_length: int = 50) -> tuple[list, dict]:
    """基础质量过滤"""
    passed, stats = [], {"total": len(docs), "filtered": {}}
    for doc in docs:
        text = doc["text"]
        if len(text) < min_length:
            stats["filtered"].setdefault("too_short", 0)
            stats["filtered"]["too_short"] += 1
            continue
        if len(set(text)) / max(len(text), 1) < 0.05:
            stats["filtered"].setdefault("low_entropy", 0)
            stats["filtered"]["low_entropy"] += 1
            continue
        passed.append(doc)
    stats["passed"] = len(passed)
    return passed, stats

# ====== 管线 Step 2：精确去重 ======
def exact_dedup(docs: list[dict]) -> tuple[list, dict]:
    """基于 SHA256 的精确去重"""
    seen_hashes = set()
    deduped, stats = [], {"total": len(docs), "duplicates": 0}
    for doc in docs:
        text_hash = hashlib.sha256(doc["text"].encode()).hexdigest()
        if text_hash in seen_hashes:
            stats["duplicates"] += 1
            continue
        seen_hashes.add(text_hash)
        deduped.append(doc)
    stats["passed"] = len(deduped)
    return deduped, stats

# ====== 管线 Step 3：统计报告 ======
def generate_report(docs: list[dict], stage_stats: list[dict]) -> dict:
    """生成处理报告"""
    lengths = [len(d["text"]) for d in docs]
    return {
        "final_count": len(docs),
        "avg_length": round(sum(lengths) / max(len(lengths), 1), 1),
        "total_chars": sum(lengths),
        "pipeline_stages": stage_stats,
    }

# ====== 运行管线 ======
print("=" * 50)
print("迷你数据处理管线")
print("=" * 50)

all_stats = []

# Step 1
filtered, s1 = basic_filter(raw_documents)
all_stats.append({"stage": "basic_filter", **s1})
print(f"\n[Step 1] 基础过滤: {s1['total']} → {s1['passed']} (过滤 {s1['filtered']})")

# Step 2
deduped, s2 = exact_dedup(filtered)
all_stats.append({"stage": "exact_dedup", **s2})
print(f"[Step 2] 精确去重: {s2['total']} → {s2['passed']} (重复 {s2['duplicates']})")

# 报告
report = generate_report(deduped, all_stats)
print(f"\n[Report] 最终: {report['final_count']} 文档, "
      f"平均长度 {report['avg_length']} 字符, "
      f"总计 {report['total_chars']} 字符")

练习扩展：

  1. 添加 MinHash 模糊去重步骤
  2. 添加语言识别过滤（使用 langdetect 或 fasttext）
  3. 为每个阶段添加质量门控（如果过滤率异常则报警）
  4. 生成数据血缘记录（记录每条文档经过了哪些处理）

本章要点回顾

  1. datatrove 和 NeMo-Curator 是当前最主流的开源框架——前者适合 CPU 集群，后者利用 GPU 加速
  2. PB 级数据处理的成本不可忽视：存储是大头（~$700K/月/30PB），一次完整管线运行约 $150-250K
  3. 分布式去重是工程最难的环节——Union-Find 需要全局状态，是并行化的瓶颈
  4. 数据版本管理不是奢侈品——至少用 manifest 文件记录每个版本的文件列表和处理参数
  5. 数据血缘追踪：每条数据的完整处理历史应该可查询，否则无法复现和调试
  6. CI/CD for Data：数据管线应该像代码一样有自动化测试、质量门控和告警机制
  7. 80% 的时间花在调试和重试——checkpoint、日志和重试机制是生产级管线的必备