[Tracking] Reduce peak memory usage across Data-Juicer operators / 降低算子峰值内存

[macroguo-ghy]

English

Motivation

We use Data-Juicer extensively in internal, production-scale data-processing workflows. In practice, long-tail samples, highly repetitive inputs, and worker concurrency can amplify an operator's transient allocations: even when the dataset fits comfortably overall, a single sample or processing stage can cause a sharp resident-set-size (RSS) spike and an out-of-memory (OOM) failure.

We would like to systematically identify and reduce avoidable peak memory usage in Data-Juicer operators, and contribute the generally useful improvements back to the community. This issue is a long-running tracker and discussion entry point; each optimization will remain a small, independently reviewable issue or PR.

Goals

• Reduce operator-local peak RSS and, where applicable, the rate at which memory grows with input size.
• Preserve output semantics, public APIs, error behavior, caching, and integration behavior unless a change is explicitly discussed.
• Avoid introducing arbitrary input, token, worker, or concurrency limits as a substitute for fixing avoidable allocations.
• Keep throughput regressions measurable and small; prefer changes that are neutral or beneficial to latency.
• Use focused changes such as streaming aggregation, bounded block processing, incremental reduction, and avoiding unnecessary materialization.

The initial focus is CPU-side transient memory allocated inside individual operators. Executor-wide scheduling, distributed-runtime memory, model/GPU memory, and dataset storage have different failure modes and should be investigated separately, while related issues may still be linked here.

Evidence expected for each optimization

Each child issue or PR should provide evidence proportional to its risk, ideally including:

1. A reproducible comparison against the implementation on the relevant main revision, using identical inputs and environment.
2. Peak RSS at multiple input sizes, preferably measured in fresh processes with repeated runs, plus the memory-growth trend rather than only one favorable point.
3. Throughput or latency measurements, including tail latency where it is meaningful.
4. Exact or reference-based differential tests, existing operator tests, and adversarial boundary cases relevant to the optimization.
5. A clear description of the workload, environment, measurement method, and limitations. Reproducible benchmark code may live on a separate branch when it is not intended to run in normal CI.

An optimization does not have to win on every workload, but its applicable workload and trade-offs should be explicit.

Tracking

Operator	Optimization	Status	Issue / PR	Evidence
character_repetition_filter and word_repetition_filter	Stream n-gram frequency counting instead of retaining an occurrence-sized list	Merged	#1028 / #1029	At 2.4M occurrences, peak RSS decreased by 41.2% and 52.2%; RSS growth slopes decreased by 95.7% and 93.9%; median throughput improved by 6.4% and 5.4%
document_minhash_deduplicator	Reduce token-by-permutation matrices in bounded blocks and fold exact minima	Merged	#1034 / #1035	At 131,072 unique shingles and 256 permutations, peak RSS decreased from 768.0 MiB to 271.9 MiB (-64.6%), the RSS growth slope decreased by 99.0%, and median throughput was 1.008x
token_num_filter	Tokenize at most 128 uncached rows per call and retain only scalar counts between calls	In review	#1039 / #1040	With the default Hugging Face tokenizer at 1,000 rows and 515-516 tokens per row, peak RSS decreased from 639.2 MiB to 562.0 MiB (-12.1%), Python allocation peak decreased by 86.6%, and median throughput was 1.102x

More candidates will be added only after their community baseline, semantics, and reproducible workload have been confirmed.

Linking convention

• A focused child issue or PR should use Part of #<tracking-issue>.
• If a PR resolves a focused child issue, use both Fixes #<child-issue> and Part of #<tracking-issue>.
• Child PRs should not use Fixes or Closes for this tracking issue. It should remain open until the overall effort is intentionally concluded.

Feedback on the scope, evidence standard, candidate operators, and benchmark methodology is welcome.

---

中文

背景与动机

我们在内部生产规模的数据处理流程中大量使用 Data-Juicer。实践中，长尾超长样本、高重复输入以及 worker 并发会放大算子的临时内存分配：即使整个数据集的总体规模可以正常处理，单条样本或某个处理阶段仍可能造成常驻内存（RSS）突然升高并触发 OOM。

我们希望系统性识别并降低 Data-Juicer 算子中可避免的峰值内存占用，并将具有通用价值的改进回馈给社区。这个 issue 是一个长期跟踪项和讨论入口；每项优化仍会拆分成小型、可独立审查的 issue 或 PR。

目标

• 降低算子内部的峰值 RSS，并在适用时降低内存随输入规模增长的速度。
• 除非经过明确讨论，否则保持输出语义、公开 API、异常行为、缓存行为和集成行为不变。
• 不用任意限制输入长度、token 数、worker 数或并发度的方式替代对可避免内存分配的修复。
• 对吞吐或延迟变化进行量化，尽量将回退控制在较小范围，优先选择性能持平或更优的方案。
• 优先采用流式聚合、有界分块、增量归约和避免不必要物化等局部且明确的优化方式。

初始范围聚焦于单个算子内部的 CPU 临时内存。执行器整体调度、分布式运行时内存、模型/GPU 内存以及数据集存储具有不同的故障机制，建议分别调查；相关 issue 仍可以在此关联。

每项优化期望提供的证据

每个子 issue 或 PR 应根据改动风险提供相应证据，理想情况下包括：

1. 在对应 main 版本的原实现上进行可复现对比，并保证输入和运行环境一致。
2. 在多个输入规模下测量峰值 RSS；优先使用全新进程、重复运行，并展示内存增长趋势，而不是只提供单个有利数据点。
3. 提供吞吐或延迟数据；在有意义时同时报告尾延迟。
4. 提供精确或基于参考实现的差分测试、已有算子测试，以及与本次优化相关的对抗性边界测试。
5. 明确说明负载、环境、测量方法和局限。若 benchmark 不适合在常规 CI 中运行，可将可复现代码放在独立分支。

一项优化不必在所有负载上都取得收益，但必须明确适用场景和权衡。

进度跟踪

算子	优化方式	状态	Issue / PR	证据
character_repetition_filter 和 word_repetition_filter	流式统计 n-gram 频次，避免保留与出现次数等大的列表	已合并	#1028 / #1029	在 240 万次出现时，峰值 RSS 分别降低 41.2% 和 52.2%；RSS 增长斜率分别降低 95.7% 和 93.9%；中位吞吐分别提升 6.4% 和 5.4%
document_minhash_deduplicator	对 token×permutation 矩阵进行有界分块，并精确归并各块最小值	已合并	#1034 / #1035	在 131,072 个唯一 shingle、256 个 permutation 时，峰值 RSS 从 768.0 MiB 降至 271.9 MiB（-64.6%），RSS 增长斜率降低 99.0%，中位吞吐为 1.008x
token_num_filter	每次最多处理 128 条未缓存文本，并在批次之间仅保留整数 count	审查中	#1039 / #1040	使用默认 Hugging Face tokenizer、1,000 行且每行 515-516 tokens 时，峰值 RSS 从 639.2 MiB 降至 562.0 MiB（-12.1%），Python allocation peak 降低 86.6%，中位吞吐为 1.102x

只有在社区版本基线、算子语义和可复现负载得到确认后，才会继续加入其他候选项。

关联约定

• 独立的子 issue 或 PR 使用 Part of #<tracking-issue>。
• 如果 PR 会解决一个独立子 issue，同时使用 Fixes #<child-issue> 和 Part of #<tracking-issue>。
• 子 PR 不应对本 tracking issue 使用 Fixes 或 Closes；在整体工作被明确结束前，它应保持打开状态。

欢迎社区对范围、证据标准、候选算子和 benchmark 方法提出建议。