用途
DataTrove 是一个用于在大规模场景下处理、过滤和去重文本数据的库。它把文本数据处理从脚本的混乱中解放出来，提供一组与平台无关、可自定义的流水线处理模块。其处理流水线可在本地直接运行，也可在 Slurm 集群上运行。较低的内存占用和多步骤设计，使其适合大规模负载，例如处理大语言模型的训练数据。
主要能力
DataTrove 提供一组预置的常用处理模块，并提供便于扩展自定义功能的框架。流水线中的每个模块以 datatrove Document 格式处理数据。Document 包含：text（每个样本的实际文本内容）、id（该样本的唯一字符串 ID）、metadata（可存放任意附加信息的字典）。每个流水线模块接收 Document 生成器作为输入，并返回另一个 Document 生成器。
流水线模块类型包括：readers（从不同格式读取数据并生成 Document）、writers（将 Document 以不同格式保存到磁盘或云端）、extractors（从原始格式如网页 HTML 中提取文本内容）、filters（按特定规则或条件过滤移除部分 Document）、stats（收集数据集统计信息）、tokens（对数据分词或统计 token 数）、dedup（去重）。
使用方式
安装需要 Python 3.10+。可通过重复 --extra 组合不同功能，例如 uv sync --extra processing --extra s3。可用组合包括：all 安装全部；io 读取 warc/arc/wet 文件及 arrow/parquet、Optimized-parquet 格式的依赖；processing 用于文本提取、过滤和分词；s3 支持 s3；cli 提供命令行工具；ray 提供分布式计算引擎；inference 用于 LLM 推理流水线；decont 用于结合 lighteval 做去污染；multilingual 用于多语言文本处理。例如 all 安装：uv sync --extra all。
一个 pipeline 是待执行的处理步骤列表（读取数据、过滤、写入磁盘等）；executor 在给定执行环境（slurm、多 CPU 机器等）上运行特定 pipeline；job 是在给定 executor 上执行一个 pipeline；task 是 job 的组成部分，用于并行执行，通常让每个 task 处理一个数据分片，DataTrove 会记录哪些 task 已完成，重新启动时只运行未完成的 task；file 是单个输入文件（.json、.csv 等），每个文件由单个 task 处理，DataTrove 不会自动将一个文件拆成多部分，因此要充分利用并行，应使用多个中等大小的文件而非单个大文件；shard 是一组输入数据（通常是一组 file），会分配给特定 task，每个 task 处理不同的、不重叠的数据分片；worker 是一次执行一个 task 的计算资源，例如有 50 个 CPU 核心，可以运行 workers=50 的 LocalPipelineExecutor，同时执行 50 个 task。
task 数量决定可并行程度以及每个处理单元耗时。task 数少（每个 task 处理的文件多）时若失败则需从头重跑；task 数多且每个较小时，单个失败 task 重跑耗时更少。若 tasks 大于 files，部分 task 不会处理任何数据，因此通常不必把 tasks 设为大于 files。
执行器包括：LocalPipelineExecutor（在本地机器上启动流水线，选项有 tasks 总任务数、workers 同时运行的任务数、start_method 生成多进程 Pool 的方法）；SlurmPipelineExecutor（在 slurm 集群上启动流水线，使用 slurm job arrays 管理和分组任务）；RayPipelineExecutor（在 ray 集群上启动流水线，使用 ray tasks 并行执行）；JobsPipelineExecutor（实验性，可能在无通知情况下变更或移除，在 Hugging Face Jobs 上运行流水线）。
调用执行器的 run 方法即可执行其 pipeline。DataTrove 通过在 ${logging_dir}/completions 文件夹中创建标记（空文件）来记录哪些 task 已成功完成。job 结束后若有 task 失败，可重新启动完全相同的执行器，DataTrove 会检查并只运行之前未完成的 task。若因部分 task 失败而重启流水线，不要更改总 task 数，否则会影响输入文件的分片分布。
路径方面，DataTrove 通过 fsspec 支持多种输入/输出源。向 datatrove 模块提供路径（如 input_folder、logging_dir、data_folder 等参数）的方式有：str（最简单，传单个字符串，例如 /home/user/mydir、s3://mybucket/myinputdata、hf://buckets/myorg/my-bucket/raw/、hf://datasets/allenai/c4/en/）；(str, fsspec filesystem instance)（字符串路径和已完全初始化的文件系统对象）；(str, dict)（字符串路径和用于初始化 fs 的选项字典）；DataFolder（直接初始化 DataFolder 对象并传入）。这些参数组合由 get_datafolder 解析。
日志消息支持着色。默认情况下，控制台消息自动检测着色，日志文件（logs/task_XXXXX.log）禁用着色。可通过环境变量显式启用或禁用：DATATROVE_COLORIZE_LOGS 设为 "1" 为控制台日志消息添加 ANSI 颜色，设为 "0" 禁用着色；DATATROVE_COLORIZE_LOG_FILES 设为 "1" 为保存到 logs/task_XXXXX.log 的日志消息添加 ANSI 颜色。
示例与文档主题
资料中列出的示例包括：fineweb.py（完整复现 FineWeb 数据集）；process_common_crawl_dump.py（读取 commoncrawl warc 文件、提取文本内容、过滤并将结果保存到 s3 的完整流水线，运行于 slurm）；tokenize_c4.py（直接从 huggingface hub 读取数据，用 gpt2 分词器对 C4 数据集英文部分分词）；estimate_tokens.py（估算大型 HF 数据集的总 token 数）；smol_data.py（为多个大型 Hugging Face 数据集构建约 100B token 子集、50-30-20 混合以及打乱变体）；minhash_deduplication.py（对文本数据运行 minhash 去重的完整流水线）；sentence_deduplication.py（运行句子级精确去重）；exact_substrings.py（运行 ExactSubstr，需要指定仓库）；finephrase.py（用多个提示模板大规模生成合成数据集的独立示例）。
资料中列出的文档主题包括：快速开始示例、DataTrove Document、流水线模块类型、完整流水线、执行器（LocalPipelineExecutor、SlurmPipelineExecutor、RayPipelineExecutor、JobsPipelineExecutor）、DataFolder / 路径、实践指南（读取数据、合成数据生成、自定义 rollouts、即用型生成脚本、高级配置、进度监控、基准测试、提取文本、过滤数据、保存数据、去重数据、摘要统计）、自定义模块（简单数据、自定义函数、自定义模块）。
适用场景与限制
适用场景主要是大规模文本数据处理、过滤与去重，例如处理大语言模型训练数据、读取和提取 commoncrawl 数据、对数据集分词或估算 token 数、生成合成数据集等。限制方面：需要 Python 3.10+；处理大规模任务时合理设置 tasks 数量很重要；若因失败重启，不应更改总 task 数；每个文件由单个 task 处理，不会自动拆分，因此要充分利用并行需使用多个中等大小的文件；JobsPipelineExecutor 为实验性功能，可能变更或移除。其他未在资料中说明的信息暂未提供。
