基于 MinHash-LSH 的大规模文本去重方案

minhash_lsh

将输入文本分词后生成 MinHash 签名，并按 bands 划分生成对应的哈希值列表。

命令格式

minhash_lsh(
  tokens: ARRAY<STRING>,
  perms_a: ARRAY<BIGINT>,
  perms_b: ARRAY<BIGINT>,
  hash_ranges: ARRAY<INT>,
  ngram_size: INT,
  min_length: INT
)

参数说明

返回值

• 类型：ARRAY<STRING>

• 含义：每个元素为对应 band 的十六进制哈希字符串，共$$B$$个。

  输出示例：["a1b2c3", "d4e5f6", ...]

build_lsh_edges

对落入同一 LSH 桶的一组文档 ID，基于“最小节点连接”策略生成边集，用于后续图连通分量分析以聚类重复文档。

命令格式

build_lsh_edges(doc_ids: ARRAY<BIGINT>)

参数说明

返回值：

• 类型：ARRAY<STRUCT<src: LONG, dst: LONG>>

• 含义：以桶内最小 ID 为源节点，连接其余所有节点所形成的边集合。

  示例逻辑：桶内 ID 为 [1003, 1001, 1005] → 排序后取最小 1001，生成边 (1001,1003) 和 (1001,1005)

支持版本

仅以下引擎版本支持本文操作：

• esr-4.x：esr-4.1.1及之后版本。

• esr-3.x：esr-3.1.1及之后版本。

• esr-2.x：esr-2.5.1及之后版本。

步骤一：准备数据

1. 下载数据集。

   sample/10BT开源数据集的完整大小为28.5GB，请根据实际情况进行下载。本文示例仅使用该数据集的部分数据，大小为2.15GB。

2. 上传数据集至OSS。

   使用控制台或 CLI 将 Parquet 文件上传至指定路径，详见简单上传。

3. 查看上传数据。

   1. 在 Spark 环境中创建临时视图，验证数据是否可正常读取。

      CREATE OR REPLACE TEMPORARY VIEW
        temp_source USING parquet OPTIONS (path 'oss://<bucket>/fineweb-edu/sample/10BT/')
      ;

   2. 查看数据，总条数727000。

步骤二：编写脚本

保存为 MinHash.py，完整代码如下（参数说明见后表）：

import re
from typing import List
from typing import Tuple

import pyspark
import numpy as np
import numpy.typing as npt
from graphframes import GraphFrame
from pyspark.sql import SparkSession
from pyspark.sql import functions as sf
from scipy.integrate import quad as integrate

RNG = np.random.RandomState(42)
SPLIT_PATTERN = re.compile(r"[\s\xA0]+")  # 设置文本分割符
DTYPE = np.uint32
MAX_HASH = 4_294_967_295  # maximum 32-bit unsigned integer
MOD_PRIME = 4_294_967_291  # maximum 32-bit prime number

input_files = "oss://<bucket>/fineweb-edu/sample/10BT/*.parquet"  # 设置需要去重的数据文件
output_path = "oss://<bucket>/fineweb-edu/output/10BT"  # 设置输出路径
checkpoint_dir = "oss://<bucket>/fineweb-edu/checkpoints"  # 设置checkpoint，用于图连通分量计算时存储中间结果
threshold = 0.8  # 设置 threshold 参数
num_perm = 256  # 设置排列的数量
ngram_size = 5  # 设置 n_gram 大小
min_length = ngram_size  # 设置 n_gram 最小长度，通常与 ngram_size 相同
text_column = "text"  # 设置去重文本列名
index_column = "__id__"

# 设置固定默认并发度，确保图连通分量计算结果稳定性
conf = (
    pyspark.SparkConf()
    .set("spark.default.parallelism", "200")
)

spark = SparkSession.Builder() \
    .appName("MinHashLSH") \
    .config(conf=conf) \
    .enableHiveSupport() \
    .getOrCreate()

spark.sparkContext.setCheckpointDir(checkpoint_dir)
# 自动计算最优 Band 参数 (B, R)
def optimal_param(
        threshold: float,
        num_perm: int,
        false_positive_weight: float = 0.5,
        false_negative_weight: float = 0.5
):
    def false_positive_area(threshold: float, b: int, r: int):
        a, _ = integrate(lambda s: 1 - (1 - s**r)**b, 0.0, threshold)
        return a

    def false_negative_area(threshold: float, b: int, r: int):
        a, _ = integrate(lambda s: 1 - (1 - (1 - s**r)**b), threshold, 1.0)
        return a

    min_error = float("inf")
    opt = (0, 0)
    for b in range(1, num_perm + 1):
        max_r = int(num_perm / b)
        for r in range(1, max_r + 1):
            fp = false_positive_area(threshold, b, r)
            fn = false_negative_area(threshold, b, r)
            error = fp * false_positive_weight + fn * false_negative_weight
            if error < min_error:
                min_error = error
                opt = (b, r)
    return opt


# B (Bands), R (Rows per Band), B × R = num_perm
# B 和 R 会将 MinHash 签名矩阵分成 B 个 bands，每个 band 包含 R 行，可以自行指定该参数，也可以通过 threshold 和 num_perm 进行计算
B, R = optimal_param(threshold, num_perm)

HASH_RANGES_SLICE: List[int] = [i * R for i in range(B + 1)]
PERMUTATIONS: Tuple[npt.NDArray[DTYPE], npt.NDArray[DTYPE]] = (
    RNG.randint(1, MOD_PRIME, size=(num_perm,), dtype=DTYPE),
    RNG.randint(0, MOD_PRIME, size=(num_perm,), dtype=DTYPE),
)

# 若数据本身具有主键 id 列，可以直接使用主键作为 index_column，无需使用 monotonically_increasing_id 生成 index
# .withColumn(index_column, "主键col")
df = spark.read.parquet(input_files) \
    .withColumn(index_column, sf.monotonically_increasing_id())

a, b = PERMUTATIONS

# 使用 minhash_lsh 函数求解 band_hash list，并通过 explode 将 list 展开为 band_idx 和 band_hash
hash_df = df \
    .select(index_column,
            sf.split(
                sf.lower(text_column),
                pattern=SPLIT_PATTERN.pattern).alias("tokens")) \
    .select(index_column,
            sf.minhash_lsh(
                "tokens",
                a.tolist(),
                b.tolist(),
                HASH_RANGES_SLICE,
                ngram_size,
                min_length).alias("hashes")) \
    .select(index_column, sf.posexplode("hashes").alias("band_idx", "band_hash"))

# 使用 build_lsh_edges 函数对聚合后属于同一个 LSH 桶的 band_idx 基于 “最小节点连接” 策略生成节点之间的边
edges_df = hash_df.groupBy("band_idx", "band_hash") \
    .agg(sf.count(index_column).alias("cnt"), sf.collect_list(index_column).alias("doc_ids")) \
    .filter(sf.col("cnt") > 1) \
    .select(sf.build_lsh_edges("doc_ids").alias("edges")) \
    .select(sf.explode("edges").alias("edge")) \
    .selectExpr("edge.src as src", "edge.dst as dst") \
    .persist(pyspark.StorageLevel.DISK_ONLY)

# 计算出所有边的顶点
vertices_df = edges_df.select(sf.col("src").alias("id")) \
    .union(edges_df.select(sf.col("dst").alias("id"))) \
    .distinct() \
    .repartition(4096) \
    .persist(pyspark.StorageLevel.MEMORY_AND_DISK_DESER)

assignment = GraphFrame(vertices_df, edges_df).connectedComponents()
# 保留每个连通分量中 ID 最小的代表文档
df = df.join(assignment.select(sf.col("id").alias(index_column), sf.col("component").alias("__component__")),
             on=index_column, how="left") \
    .filter(sf.col("__component__").isNull() | (sf.col("__component__") == sf.col(index_column))) \
    .drop("__component__")
    
# 输出去重结果
df.write.parquet(output_path, mode="overwrite", compression="snappy")

# 清理缓存
edges_df.unpersist()
vertices_df.unpersist()

参数说明表

步骤三：上传文件

1. 进入资源上传页面。

   1. 登录E-MapReduce控制台。

   2. 在左侧导航栏，选择EMR Serverless > Spark。

   3. 在Spark页面，单击目标工作空间名称。

   4. 在EMR Serverless Spark页面，单击左侧导航栏中的文件管理。

2. 在文件管理页面，单击上传文件。

3. 在上传文件对话框中，单击待上传文件区域选择MinHash.py，或直接拖拽MinHash.py到待上传文件区域。

步骤四：创建并运行批任务

在完成脚本开发与上传后，您需要在 Serverless Spark 环境中创建一个批处理任务来执行该去重作业。

1. 在EMR Serverless Spark页面，单击左侧的数据开发。

2. 在开发目录页签下，单击图标。

3. 在弹出的对话框中，输入名称，类型使用批任务 > PySpark，单击确定。

4. 在右上角选择队列。

   添加队列的具体操作，请参见管理资源队列。

5. 在新建的开发页签中，配置以下信息，其余参数无需配置，然后单击运行。

   参数	说明
   主Python资源	选择前一个步骤中在文件管理页面上传的Python文件。本文示例是MinHash.py。
   引擎版本	esr-2.8.0/esr-3.4.0/esr-4.4.0及以上版本。
   资源配置	建议按照4CPU:16GB的比例进行配置。例如，spark.executor.cores=4，spark.executor.memory=14GB，spark.executor.memoryOverhead=2GB。
   Spark配置	spark.rdd.ensureConfigConsistency true：必填项，必须设置为 true。 执行效率优化参数： spark.sql.shuffle.partitions 1000：建议 1TB 数据以下配置为 1000 即可，每增加 1TB 数据可增加 1000。 spark.sql.files.maxPartitionBytes 256MB：控制读取阶段分区大小，避免小文件过多，通常配置为 128/256MB 即可。

6. 运行任务后，在下方的运行记录区域，单击任务操作列的详情，监控任务执行。

步骤五：验证结果

任务成功完成后，验证去重结果。

1. 在 Spark SQL 环境中创建临时视图以加载去重后的数据。

   CREATE OR REPLACE TEMPORARY VIEW
     temp_target USING parquet OPTIONS (path 'oss://<bucket>/fineweb-edu/output/10BT')
   ;

2. 查看数据，总条数724809（原 727,000 → 去除 2,191 条重复项）。

性能调优建议

为在不同数据规模和业务目标下平衡去重的精度、召回率与执行效率，建议从以下三个层面进行系统性调优。