Data-Juicer Ray 分布式 C++ 加速 MinHash LSH 去重算子 ray_bts_minhash_cpp_deduplicator 深度解析

本文以 Data-Juicer 开源仓库中 docs/operators/deduplicator/ray_bts_minhash_cpp_deduplicator.md 为核心骨架，结合其 Python 实现、C++ 内核 minhash.cpp 与 Cython 分词模块 tokenize.pyx，完整讲解这一面向大规模文本数据的近似去重算子的原理、全部参数、内部执行流程与实战配置方法。读完本文，你将掌握如何在 Ray 集群上用 ray_bts_minhash_cpp_deduplicator 以 C++ 加速的方式做文档级近似去重，并理解 MinHash LSH、BTS 分布式并查集、band/row 自动寻优等底层机制。

一、算子概览：它是什么、解决什么问题

ray_bts_minhash_cpp_deduplicator 是 Data-Juicer 中一个在 Ray 分布式模式下运行、并以 C++ 加速的 MinHash LSH 文档级近似去重器，算子类型（Type）为 deduplicator，标签（Tags）为 cpu, text，注册名 OP_NAME 为 "ray_bts_minhash_cpp_deduplicator"，其类名为 RayBTSMinhashCppDeduplicator（见 ray_bts_minhash_cpp_deduplicator.py）。

它的核心任务是：从数据集中识别并移除"近似重复"（near-duplicate）的文本样本。与精确去重（如哈希完全一致才判重）不同，近似去重要处理的是"改写、换词、增删少量片段"之后语义仍高度相似的文本——例如同一篇新闻被不同媒体转载改写、同一篇文章被截断成不同版本等。这类数据在构建大模型预训练语料时会造成严重的冗余与过拟合风险。

从源码角度看，该算子的关键特征可以归纳为三点：

1. 分布式执行：类声明 _batched_op = True，_supported_exec_modes = ("ray", "ray_partitioned")，只能在 Ray 执行模式下运行（ray_bts_minhash_cpp_deduplicator.py）。
2. C++ 加速：文档明确指出"与 ray_bts_minhash_deduplicator 相同，但其分词和 MinHash 签名计算使用 C++ 实现，以提升性能"。
3. BTS 分布式并查集：算子名中的 BTS（Balanced Tree Structured Union-Find）对应一个带负载均衡的分布式并查集实现，用于在分布式环境中把相似样本高效聚簇。

二、核心原理：MinHash + LSH + 并查集

2.1 从 shingling 到 Jaccard 相似度

算子的判重依据是 Jaccard 相似度。给定两个样本文本 A、B，将其各自转换为 token 集合后，Jaccard 相似度定义为：

J(A, B) = |A ∩ B| / |A ∪ B|

当两个样本的 Jaccard 相似度 大于等于用户指定的阈值 jaccard_threshold 时，它们被判定为近似重复，去重后只保留其中一个样本。

文本到 token 集合的转换分为两步（这一步由 Cython/C++ 完成）：

• 分词（tokenization）：按所选分词方法把文本切成 token；
• shingling（滑窗拼接）：按 window_size 把相邻 token 拼接成 n-gram 窗口（shingle），形成样本的 shingle 集合。

以 space 分词为例，代码先调用 split_on_whitespace 切词，再用 n_grams 按窗口拼接：tokens<a href="https://link.gitcode.com/i/ce1bc58e8273b1a1cb5d292deb92ee26" target="_blank">i:i+window_size] 连接成一个 bytes 窗口并去重（见 [tokenize.pyx）。shingle 集合即"文本指纹"的原料。

2.2 MinHash 签名：把大集合压缩成定长指纹

直接对海量文本两两计算 Jaccard 相似度代价极高（O(N²)）。MinHash 技巧在于：对 shingle 集合施加若干随机排列（permutation），取每个排列下的最小哈希值，得到的 num_permutations 个最小值构成该样本的签名向量。根据 MinHash 理论，两个样本签名向量中对应位置相等的比例，恰好是无偏估计两个集合的 Jaccard 相似度。

在 C++ 内核中，每个 token 先被 simple_hash 映射为 32 位哈希（基于经典的 djb2 变体，初始值 5381，见 minhash.cpp），随后对每个排列计算 (h * perm_a<a href="https://link.gitcode.com/i/db0cf851e0b680c8b03c77134025efa8" target="_blank">i] + perm_b[i]) >> 32 并取逐 token 最小值（[minhash.cpp）。排列参数 perm_a、perm_b 由 np.random.RandomState(seed=42) 固定种子生成，保证结果可复现（ray_bts_minhash_cpp_deduplicator.py）。

2.3 LSH：把签名分 band，用"必中"换取"大概率命中"

如果直接比较完整签名，仍需两两比对。LSH（局部敏感哈希）的做法是：把 num_permutations 个哈希值均匀切成 num_bands 个 band，每个 band 含 num_rows_per_band 行（即 num_bands × num_rows_per_band = num_permutations）。每个 band 的哈希序列拼成一个 band 指纹：

• 两个样本在任意一个 band 上指纹完全相同，就会被放进同一个 hash bucket；
• 哈希桶内使用并查集把它们合并为同一簇。

分 band 后，相似度越高的样本对在同一 band 上"撞桶"的概率越高（相似度 s 的两个样本，单 band 命中概率为 1 - (1 - s^r)^b）。LSH 以一定的假阳性（把不同样本误判为相似）换取远低于 O(N²) 的比较开销——只有撞桶的样本才需要被并查集处理。

2.4 BTS 分布式并查集：跨节点聚簇

所有样本（按 uid 编号）先映射到若干分布式哈希表（hash table），每个哈希桶内出现多个 uid 时调用 union_list 将它们合并；随后由分布在各节点上的 BTSUnionFind actor 通过 edge buffer 交换边信息 + rebalancing（重平衡） 迭代收敛，最终每个簇中编号最小（最先出现）的样本被保留，其余视为重复样本被过滤（ray_bts_minhash_cpp_deduplicator.py）。这一分布式并查集参考了 BTS Union-Find 相关论文方法，原版 Python 实现在 ray_bts_minhash_deduplicator.py 的类注释中有明确出处说明。

三、参数配置全解析

以下参数表完整继承自原文档，并结合源码补充了类型约束、取值范围与内部影响说明：

参数名	类型	默认值	说明
tokenization	str	'space'	样本文本的分词方法，取值必须是 [space, punctuation, character, sentencepiece] 之一。建议：类英文语言用 'space'，类中文语言用 'character'，多语言场景用 'sentencepiece'。使用 'sentencepiece' 时必须通过 tokenizer_model 提供模型路径。
window_size	Annotated[int, Gt(gt=0)]	5	shingling 的滑窗大小，即每个 shingle 包含的 token 数。
lowercase	bool	True	计算 minhash 前是否先将文本转为小写。
ignore_pattern	Optional[str]	None	计算 minhash 时是否忽略匹配特定正则模式的子串。注意：若 ignore_pattern 包含标点，punctuation 分词将失效（源码会在该组合下打出 logger.warning）。
tokenizer_model	Optional[str]	None	sentencepiece 模型路径，仅 sentencepiece 分词需要；源码规定此时若为 None 会直接抛出 ValueError。
num_permutations	Annotated[int, Gt(gt=0)]	256	MinHash 计算中的排列（permutation）数量，即签名长度。
jaccard_threshold	Annotated[float, Ge(ge=0), Le(le=1)]	0.7	近似重复检测的最小 Jaccard 相似度阈值。两个样本文本的 Jaccard 相似度 ≥ 该阈值时，它们被视为相似样本，去重后只保留其中一个。
num_bands	Optional[Annotated[int, Gt(gt=0)]]	None	LSH 中的 band 数量。默认 None，由最优参数计算算法自动确定——该算法通过最小化"假阳性概率与假阴性概率的加权和"来求最优 (bands, rows)。
num_rows_per_band	Optional[Annotated[int, Gt(gt=0)]]	None	每个 band 的行数。默认 None，同上由最优参数计算算法自动确定。
union_find_parallel_num	Union[int, str]	'auto'	分布式并查集的并行度。'auto' 时取 cpu_num // 2（CPU 数一半）。
union_threshold	Optional[int]	256	哈希桶内 uid 数量超过该阈值时，先做局部合并再继续累积，用于控制内存与合并粒度。
max_pending_edge_buffer_task	Optional[int]	20	每轮从远端 edge buffer 拉取边任务时允许的最大在途任务数（配合 ray.wait 做流式控制）。
num_edge_buffer_task_returns	Optional[int]	10	每次 ray.wait 返回的就绪任务个数。
max_pending_filter_tasks	Optional[int]	20	过滤阶段允许的最大在途 remain_mask 任务数。
num_filter_task_returns	Optional[int]	10	过滤阶段每次 ray.wait 返回的就绪任务个数。
merge_batch_size	Optional[int]	1000	合并阶段每批 ray.wait 获取的结果条数，最终会被截断为不超过 union_find_parallel_num。
args / kwargs	—	''	透传给 Deduplicator 基类的额外参数（如 text_key、work_dir 等）。

说明：上表中 union_find_parallel_num 之后的若干参数在原文档表格中说明列为空，上述解释基于 ray_bts_minhash_cpp_deduplicator.py 的构造函数签名与后续使用方式归纳，属于对实现事实的描述而非官方文档原文。

3.1 band/row 自动寻优原理

当 num_bands 或 num_rows_per_band 为 None 时，构造函数调用 document_minhash_deduplicator.py 中导出的 optimal_param 函数（document_minhash_deduplicator.py）：

if self.num_bands is None or self.num_rows_per_band is None:
    self.num_bands, self.num_rows_per_band = optimal_param(
        self.jaccard_threshold,
        self.num_permutation,
    )

optimal_param 在 1 ≤ b ≤ num_perm、1 ≤ r ≤ num_perm/b 的候选组合中暴力搜索，目标函数为：

error = FP(b, r) * false_positive_weight + FN(b, r) * false_negative_weight

其中假阳性概率与假阴性概率通过对 band 命中公式 1 - (1 - s^r)^b 在 [0, threshold] 与 [threshold, 1] 区间上做数值积分得到（权重默认各 0.5）。随后代码用 (i * num_rows_per_band, (i+1) * num_rows_per_band) 构造每个 band 的哈希区间 hash_ranges，供 C++ 内核按 band 切分签名。

3.2 分词方法在源码中的实现差异

MinhashCalculator 构造函数按 tokenization 分派不同的分词函数（ray_bts_minhash_cpp_deduplicator.py）：

分词方法	实现
space	Cython 实现 split_on_whitespace（逐字节扫描 UTF-8 串切分空白）
punctuation	用 regex.compile(r"\p{P}")（Unicode 标点类别）切分
character	逐字符切分并 UTF-8 编码
sentencepiece	通过 prepare_sentencepiece_model(tokenizer_model) 加载模型后 encode(x, out_type=str)

space 与 shingling 的 n_grams 均由 tokenize.pyx 提供，是 Cython 加速的重点之一。

四、C++ 加速内核：minhash.cpp 到底快在哪

与纯 Python 的 ray_bts_minhash_deduplicator（在 Python 层逐样本用 sha1_hash32 + numpy 批量置换求最小值）不同，C++ 版本把token 哈希、permutation 置换、band 指纹拼装整体下沉到 C++：

1. djb2 式快速哈希：simple_hash 直接对 token 字节串计算 32 位哈希（minhash.cpp），省去了 Python 层 sha1 的开销；
2. OpenMP 多线程批量计算：calc_minhash_batch_c 通过 omp_set_num_threads(num_threads) 设置线程数，用 #pragma omp for nowait 并行遍历批次内所有文档，每个线程维护本地 local_pairs，最后经 #pragma omp critical 合并（minhash.cpp）；
3. 单趟求最小：对每个 token 同时更新全部 num_permutation 个排列的最小值，避免 Python 层多次 numpy 中间数组分配；
4. band 指纹直接以 bytes 拼装：(band_index, uint32 序列) 编码成 bytes，作为哈希桶键直接返回给 Python 侧。

Python 侧通过 pybind11 暴露的 calc_minhash_batch_c 以 thread_num=4 的默认线程数调用（ray_bts_minhash_cpp_deduplicator.py）。空 token 列表的样本会被赋予 MAX_HASH 的"空指纹"，统一落入 MAX_HASH % union_find_parallel_num 对应的桶，保证空文档也能被一致处理。

此外 C++ 版本把批大小 BATCH_SIZE 设为 2048（见 ray_bts_minhash_cpp_deduplicator.py），而 Python 版为 1000（ray_bts_minhash_deduplicator.py），更大的批次粒度更有利于发挥 C++ 批量计算的吞吐。注意：本文不给出具体加速倍数等未经仓库证实的数据，以上仅是对实现方式的事实性描述。

五、执行流程：Ray 集群上的三段式流水线

run() 方法是算子的分布式执行主流程（ray_bts_minhash_cpp_deduplicator.py），整体分为三个阶段：

阶段一：MinHash 签名计算（map）

dataset.map_batches(
    MinhashCalculator,
    batch_format="pyarrow",
    zero_copy_batch=True,
    concurrency=self.min_hash_concurrency,  # cpu_num - 4 * num_nodes
    ...
).write_parquet(self.tmp_dir, ray_remote_args={"num_cpus": 0.25})

• 构造函数中按节点数量创建 IdGenerator、HashAggregator actor，全部使用 NodeAffinitySchedulingStrategy 绑定到具体 worker 节点（soft=False），保证 uid 分配与哈希聚合的位置局部性；
• MinhashCalculator.__call__ 为每个样本从 IdGenerator 领取全局唯一 uid，计算 MinHash 后将 (hash_table_id, band_hash_value, uid) 三元组推送到对应节点的 HashAggregator；
• 中间结果（含 uid 列）落盘到 work_dir/.tmp/<job_id> 临时目录下的 parquet，便于后续阶段读取。

阶段二：BTS 分布式并查集合并（merge）

merge() 启动多轮迭代（ray_bts_minhash_cpp_deduplicator.py）：

1. edge_redistribution：从所有 HashAggregator 取回本分片哈希表，对撞桶 uid 做局部 union，再按 uid // BATCH_SIZE % parallel_num 将边重新分发到各 edge buffer；
2. balanced_union_find：消费本地与远端 edge buffer 中的边做 union，随后 rebalancing() 把簇代表（parent）收敛到编号最小的样本，返回 old_parent != parent 表示是否仍有变化；
3. while any(...) 循环直到所有并查集 actor 都报告收敛，期间 communication() 负责把变更的边在分片之间传递；
4. squeeze()：生成每个 uid 分片的保留掩码 dup_uids，簇内非代表样本被标记为去重。

合并阶段通过 max_pending_edge_buffer_task / num_edge_buffer_task_returns 控制 ray.wait 的流式拉取，避免一次性堆积过多远端任务；union_threshold 则限制单桶累积长度，超过即提前 union_list 压缩。启动时还会把 DEFAULT_MAX_TASKS_IN_FLIGHT 调为 2 以配合 Ray Data 的 actor 池调度。

阶段三：过滤（filter）

dataset = ray.data.read_parquet(self.tmp_dir)
result = dataset.map_batches(
    MinhashFilter,
    batch_format="pyarrow",
    zero_copy_batch=True,
    concurrency=self.filter_concurrency,  # cpu_num * 0.75
    ...
)

MinhashFilter.__call__ 按 uid 分片查询对应 union_find_{hash_id} actor 的 remain_mask，用 bitarray 压缩掩码逐批获取保留位图，最后剔除 uid 列并 filter(mask) 输出。每个重复簇中保留的是 uid 最小（即最早出现）的样本。

六、实战：如何在配置文件中启用

6.1 最小配置示例

在 Data-Juicer 的 YAML 流程配置中，按算子注册名引用即可（text_key 指定去重依据的文本字段，work_dir 指定临时目录）：

# process.yaml（片段）
process:
  - ray_bts_minhash_cpp_deduplicator:
      text_key: text
      tokenization: space        # 英文语料推荐
      window_size: 5
      lowercase: true
      num_permutations: 256
      jaccard_threshold: 0.7     # Jaccard ≥ 0.7 判为近似重复
      work_dir: ./output/dedup_tmp

对应命令行运行（需先启动 Ray 集群，或在本地以 Ray 模式执行）：

python tools/process_data.py --config process.yaml --execution-type ray

6.2 不同语料的推荐组合

语料类型	推荐配置	依据
类英文语料	tokenization: space	文档推荐
类中文语料	tokenization: character	文档推荐
多语言混合	tokenization: sentencepiece + tokenizer_model: <模型路径>	文档推荐；且源码要求必须提供模型路径，否则抛 ValueError
需要忽略特殊模式	增加 ignore_pattern: r'\p{P}'	测试用例即采用此模式过滤标点；若与 punctuation 分词同时使用需谨慎

测试用例 test_ray_bts_minhash_deduplicator.py 展示了英文去重的标准用法：RayBTSMinhashCppDeduplicator(ignore_pattern=r'\p{P}', work_dir=work_dir)；中文用例则使用 tokenization='character', ignore_pattern=r'\p{P}'（test_ray_bts_minhash_deduplicator.py）。两个用例都验证了"输入 N 条含近似重复的文本，输出保留列表与预期完全一致"。

七、运行前提与注意事项

1. 必须运行在 Ray 模式：该算子 _supported_exec_modes = ("ray", "ray_partitioned")，不支持单机非 Ray 执行；使用前需确保已按 Data-Juicer 的 Ray 使用文档 初始化集群。
2. 临时目录：中间 parquet 写入 work_dir/.tmp/<job_id>，任务结束后会由流程清理；建议为 work_dir 预留足够磁盘空间。
3. C++ 扩展依赖：minhash（pybind11 模块）与 tokenize（Cython 模块）需在安装 Data-Juicer 时完成编译（参考 pyproject.toml 构建配置），minhash.cpp 依赖 OpenMP。
4. punctuation 分词与 ignore_pattern 的冲突：若 ignore pattern 包含标点字符，标点分词将无法正确工作，源码会打印警告（ray_bts_minhash_cpp_deduplicator.py）。
5. 阈值语义：jaccard_threshold 是"保留一个、删除其余"的边界——≥ 阈值 即判为相似；阈值越高，判重越严、保留越多，反之去重越激进。band/row 参数建议保持 None 由 optimal_param 自动寻优，除非你对 LSH 的假阳性/假阴性权衡有明确把握。
6. 大规模并行度控制：union_find_parallel_num='auto' 取 CPU // 2；MinHash 计算并发度为 CPU - 4 * 节点数，过滤并发度为 CPU * 0.75，这些数值由集群 ray.cluster_resources() 的 CPU 数动态推导，无需手工配置。

八、相关资源

• 算子源码：data_juicer/ops/deduplicator/ray_bts_minhash_cpp_deduplicator.py
• C++ 内核：data_juicer/ops/deduplicator/minhash.cpp
• Cython 分词模块：data_juicer/ops/deduplicator/tokenize.pyx
• 纯 Python 对照实现（含 BTS 出处说明与 uid 变体）：data_juicer/ops/deduplicator/ray_bts_minhash_deduplicator.py
• 单机版 MinHash 与 optimal_param 来源：data_juicer/ops/deduplicator/document_minhash_deduplicator.py
• 单元测试（含英文、中文去重断言）：tests/ops/deduplicator/test_ray_bts_minhash_deduplicator.py
• 全部算子列表：docs/operators/Operators.md
• Ray 分布式执行指南：docs/Distributed.md