E-MapReduce：基於 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 條重複項）。

效能調優建議

為在不同資料規模和營運目標下平衡去重的精度、召回率與執行效率，建議從以下三個層面進行系統性調優。