xpark.dataset.HashDedup#

class xpark.dataset.HashDedup(hash_algorithm: str = 'xxh3_128', parallel_num: int | None = None, is_file: bool = False)[source]#

Exact-hash-based deduplication for text, binary data, or files.

Computes a hash for each record and uses distributed HashSet Actors to detect duplicates. The first occurrence of each hash value is kept; subsequent duplicates are filtered out.

Parameters:

  • hash_algorithm – Hash algorithm to use. Accepts any algorithm supported by hashlib (e.g. 'md5', 'sha1', 'sha256', 'sha3_256', 'blake2b') or xxhash (e.g. 'xxh64', 'xxh3_64', 'xxh3_128'). Default: 'xxh3_128'.

  • parallel_num – Number of HashSet Actor shards. If None, it is derived automatically from the cluster CPU count (max(cpu_count // 4, 1)).

  • is_file – If True, treat each column value as a file path supported by fsspec (e.g. local paths, s3://, cos://, gs://, etc.) and hash the file contents. Useful for deduplicating audio, video, or other binary files by content. Default: False.

Example — text deduplication:

>>> from xpark.dataset import read_parquet
>>> from xpark.dataset.filters.dedup import HashDedup
>>> from xpark.dataset.expressions import col
>>>
>>> ds = read_parquet("/data/my-dataset")
>>> ds.filter(
...     HashDedup().with_column(text=col("text"))
... ).write_parquet("/data/dedup-my-dataset")

Example — video file deduplication:

>>> from xpark.dataset import read_parquet
>>> from xpark.dataset.filters.dedup import HashDedup
>>> from xpark.dataset.expressions import col
>>>
>>> # Each row has a "video_path" column pointing to a local video file.
>>> ds = read_parquet("/data/video-dataset")
>>> ds.filter(
...     HashDedup(hash_algorithm="sha256", is_file=True).with_column(text=col("video_path"))
... ).write_parquet("/data/dedup-video-dataset")

Methods

with_column(text)

with_column(text: ColumnExpr) → DedupOp[source]#