How EMR Serverless Spark Achieves 4× Faster PB‑Scale Text Deduplication

Why Text Deduplication Matters for Large‑Scale LLM Training

Training data quality directly impacts model performance; duplicated documents waste compute, cause over‑fitting, and inflate evaluation metrics. When corpus size reaches petabytes, deduplication becomes a compute‑intensive "performance black hole" that can stall for an entire night and fail under data skew.

Limitations of the Original Open‑Source Spark Setup

Compute efficiency bottleneck: Row‑based JVM execution and Python UDFs introduce heavy virtual‑function calls, object boxing, and cross‑process serialization, leading to low CPU cache utilization.

Shuffle stability issues: Heavy shuffle operations in MinHash‑LSH cause frequent timeouts and failures under skewed data.

High operational cost: Maintaining a self‑managed Spark cluster requires continuous manpower for version upgrades, resource scheduling, and fault handling.

Technical Solution in EMR Serverless Spark

After migration, the workflow relies on two core capabilities:

MinHash‑LSH built‑in functions: The algorithm is exposed as minhash_lsh (generates MinHash signatures) and build_lsh_edges (creates graph edges for LSH buckets), reducing user code by about 40%.

Fusion Engine (Spark Native Engine): A vectorized execution engine written in C++ that accelerates hash computation, removes Python UDF overhead, and provides shuffle‑stability optimizations for the compute‑storage separation architecture.

MinHash‑LSH Details

The algorithm works in two steps:

Step 1 – MinHash: Convert each document to an n‑gram set, apply multiple hash functions, and produce a fixed‑length signature (e.g., 256 bits). This signature acts as a "fingerprint" for the document.

Step 2 – LSH banding: Split the signature into several "bands"; documents that fall into the same hash bucket are likely similar and only those pairs are compared, reducing the complexity from O(n²) to near‑linear.

Serverless Spark implements these steps with the following built‑in functions:

minhash_lsh(
    tokens: ARRAY<STRING>,            -- tokenized words
    perms_a: ARRAY<BIGINT>,           -- MinHash multiplier parameters
    perms_b: ARRAY<BIGINT>,           -- MinHash additive parameters
    hash_ranges: ARRAY<INT>,          -- band boundaries
    ngram_size: INT,                  -- n‑gram size (5‑9 for long texts)
    min_length: INT                  -- minimum token length
)  -- returns ARRAY<STRING> of hex hash values per band

build_lsh_edges(doc_ids: ARRAY<BIGINT>)  -- returns ARRAY<STRUCT<src: LONG, dst: LONG>> of edges

Using these functions, the user code shrinks to a few Spark SQL expressions, for example:

hash_df = df \
    .select(index_column, split(lower(text_column), pattern).alias("tokens")) \
    .select(index_column, minhash_lsh("tokens", a.tolist(), b.tolist(), HASH_RANGES_SLICE, ngram_size, min_length).alias("hashes")) \
    .select(index_column, posexplode("hashes").alias("band_idx", "band_hash"))

edges_df = hash_df.groupBy("band_idx", "band_hash") \
    .agg(count(index_column).alias("cnt"), collect_list(index_column).alias("doc_ids")) \
    .filter(col("cnt") > 1) \
    .select(build_lsh_edges("doc_ids").alias("edges")) \
    .select(explode("edges").alias("edge")) \
    .selectExpr("edge.src as src", "edge.dst as dst")

Fusion Engine Advantages

Vectorized hash computation: MinHash signatures are generated in columnar batches, dramatically lowering per‑document latency.

No Python UDF cross‑process cost: Hash logic runs as native C++ code inside the engine, eliminating serialization overhead.

Shuffle stability: Dedicated optimizations for the compute‑storage separation architecture resolve skew‑induced timeouts.

In the customer’s production workload, these improvements yielded a 4‑5× reduction in total deduplication time.

Migration Practice – Three‑Step Process

1. Data Migration

Raw text files were moved from the previous cloud storage to Alibaba Cloud OSS. Serverless Spark supports the OSS‑HDFS protocol, allowing transparent access. Checksums ensured data integrity.

2. Code Migration

Because Spark API compatibility is preserved, the only code changes were path updates to OSS and replacement of custom Python hash logic with the built‑in minhash_lsh and build_lsh_edges functions. The overall codebase shrank by roughly 40%.

3. Resource Configuration

Serverless Spark eliminates fixed‑size clusters; executors are provisioned on‑demand (recommended 4 CPU : 16 GB memory). Key Spark settings include: spark.sql.shuffle.partitions = 1000 for data ≤ 1 TB, plus 1000 per additional TB. spark.sql.files.maxPartitionBytes = 256MB to control input split size. spark.rdd.ensureConfigConsistency = true (required). spark.executor.cores = 4, spark.executor.memory = 14GB (+2 GB overhead).

Performance Validation

Using the same MinHash‑LSH parameters (num_perm = 256, threshold = 0.8, ngram_size = 5), the new architecture reduced total deduplication time from 1–2 days to a few hours (4‑5× speedup). Shuffle failure rate dropped from frequent failures to zero, and operational effort fell from a dedicated team to near‑zero.

On the public fineweb‑edu 10 TB subset (2.15 GB, 727 k documents), Serverless Spark removed 2 191 duplicates, retaining 724 809 unique documents with high accuracy.

FAQ

Q1: Which EMR Serverless Spark versions support the MinHash‑LSH functions?

Supported in ESR 4.x (≥ 4.1.1), ESR 3.x (≥ 3.1.1) and ESR 2.x (≥ 2.5.1). Using the latest version gives the best performance.

Q2: How much code change is required when migrating?

Only data‑path updates and function replacements; overall code volume decreased by ~40%.

Q3: What data scale can Serverless Spark handle for text deduplication?

From GB to PB; for data ≤ 1 TB use 1000 shuffle partitions, adding 1000 partitions per additional TB.

Q4: How is deduplication accuracy controlled?

By tuning num_perm, threshold, and LSH band parameters; the recommended balance is num_perm=256 and threshold=0.8.

Q5: Beyond deduplication, what other AI data‑pre‑processing scenarios does Serverless Spark support?

Data cleaning, feature engineering, vector computation, multimodal processing, and direct large‑model inference via built‑in AI functions.

Conclusion

Text deduplication is a foundational step for high‑quality LLM corpora. Migrating to EMR Serverless Spark, with MinHash‑LSH built‑in functions and the Fusion Engine, delivers four‑fold speed gains, eliminates shuffle failures, and removes most operational burden, making it a compelling choice for PB‑scale data‑plus‑AI workloads.