xpark.dataset.TextFuzzyDedup#

class xpark.dataset.TextFuzzyDedup(tokenize_regex_pattern: str | None = None, shingling_window_size: int = 5, lowercase: bool = True, cjk: bool = True, jaccard_threshold: float = 0.7, minhash_num_perm: int = 256, minhash_seed: int = 42, lsh_num_bands: int | None = None, lsh_num_rows_per_band: int | None = None, union_find_parallel_num: int | None = None, union_batch_size: int = 256, union_balancing_batch_size: int = 1)#

Text fuzzy deduplication using MinHashLSH.

This class performs fuzzy deduplication on text data by: 1. Shingling: Breaking text into overlapping token sequences (shingles) 2. MinHash: Computing compact signatures for each text 3. LSH (Locality-Sensitive Hashing): Efficiently finding similar texts 4. Union-Find: Grouping duplicates using a distributed union-find algorithm

Parameters:

  • tokenize_regex_pattern – Split by space if None else apply the regex to tokenize

  • shingling_window_size – Size of shingles (tokens for space, characters for character)

  • lowercase – Whether to convert text to lowercase before processing

  • cjk – Whether to use CJK split, only available if tokenize_regex_pattern is None

  • jaccard_threshold – Jaccard similarity threshold for duplicates

  • minhash_num_perm – Number of permutations for MinHash

  • minhash_seed – Seed for random number generation (optional)

  • lsh_num_bands – Number of bands for LSH (computed if None)

  • lsh_num_rows_per_band – Number of rows per band (computed if None)

  • union_find_parallel_num – Number of union find to use for parallelization

  • union_batch_size – Batch size for union finding

  • union_balancing_batch_size – BTS batch size for union finding

Example

>>> from xpark.dataset import read_parquet, TextFuzzyDedup
>>> from xpark.dataset.expressions import col
>>>
>>> # Read parquet files with `dynamic_uid` to generate a unique ID column
>>> ds = read_parquet("/data/fineweb-edu-sample-10BT", dynamic_uid="uid")
>>>
>>> # Apply fuzzy deduplication and save results
>>> ds.filter(
...     TextFuzzyDedup().with_column(uid=col("uid"), text=col("text"))
... ).drop_columns("uid").write_parquet("/data/dedup-fineweb-edu-sample-10BT")
Inspecting duplicate groups:

Keep a reference to the TextFuzzyDedup instance and, after the filter pipeline has been executed (e.g. via take_all / write_* / materialize), call dump_groups() to collect the global duplicate clusters onto the driver. See dump_groups() for the return schema and semantics.

>>> dedup = TextFuzzyDedup()
>>> ds = read_parquet("data/", dynamic_uid="uid").filter(
...     dedup.with_column(uid=col("uid"), text=col("text"))
... )
>>> ds.write_parquet("/data/output")
>>> groups = dedup.dump_groups()
>>> # groups: {root_uid: [duplicate_uid, ...], ...}

Methods

dump_groups()

Collect the global duplicate groups onto the driver.

with_column(uid, text)

Apply the TextFuzzyDedup filter with specified columns.
dump_groups() → dict[int, list[int]]#

Collect the global duplicate groups onto the driver.

Note

Duplicate entries are not expected to reach an extreme scale, so results are merged on a single node for export. For heavier distributed post-processing, use the lower-level actor.dump_bts directly, or file an issue to request built-in support.

with_column(uid: ColumnExpr, text: ColumnExpr) → DedupOp#

Apply the TextFuzzyDedup filter with specified columns.

Parameters:

  • uid – Column expression for unique identifiers. Should be integer type. Use dynamic_uid when reading data to auto-generate unique IDs.

  • text – Column expression for text content to deduplicate. support large string / string.

Returns:

A deduplication operation that can be used with filter().

Return type:

DedupOp

Example

>>> from xpark.dataset import read_parquet, TextFuzzyDedup
>>> from xpark.dataset.expressions import col
>>> ds = read_parquet("data/", dynamic_uid="uid")
>>> ds.filter(TextFuzzyDedup().with_column(uid=col("uid"), text=col("text")))