nemo_curator.stages.deduplication.fuzzy.workflow

Module Contents

Classes

Name	Description
`FuzzyDeduplicationWorkflow`	A pipeline that performs fuzzy deduplication of a dataset.

Data

ID_GENERATOR_OUTPUT_FILENAME

API

class nemo_curator.stages.deduplication.fuzzy.workflow.FuzzyDeduplicationWorkflow(
    cache_path: str,
    output_path: str,
    input_path: str | list[str] | None = None,
    input_filetype: typing.Literal['jsonl', 'parquet'] = 'parquet',
    input_blocksize: str | int = '1GiB',
    input_file_extensions: list[str] | None = None,
    read_kwargs: dict[str, typing.Any] | None = None,
    cache_kwargs: dict[str, typing.Any] | None = None,
    write_kwargs: dict[str, typing.Any] | None = None,
    text_field: str = 'text',
    perform_removal: bool = False,
    seed: int = 42,
    char_ngrams: int = 24,
    num_bands: int = 20,
    minhashes_per_band: int = 13,
    use_64_bit_hash: bool = False,
    bands_per_iteration: int = 5,
    lsh_num_output_partitions: int | None = None,
    lsh_rmm_pool_size: int | typing.Literal['auto'] | None = 'auto',
    lsh_spill_memory_limit: int | typing.Literal['auto'] | None = 'auto',
    env_vars: dict[str, typing.Any] | None = None
)

Bases: WorkflowBase

A pipeline that performs fuzzy deduplication of a dataset. It consists of the following stages:

FilePartitioningStage Groups input files into smaller groups that can be processed in parallel.
MinHashStage Computes minhashes for the input dataset.
LSHStage Performs Locality Sensitive Hashing on the minhashes. This is a shuffle stage that involves moving data between workers.
BucketsToEdgesStage This stage converts the resulting LSH mapping of bucket ID to document ID into a graph of edges.
ConnectedComponentsStage Performs weaklyconnected components clustering on the graph represented by the edgelist.
IdentifyDuplicatesStage Generates a list of document ids to remove based on the connected components clusters/components.
Removal (Optional) Currently not implemented.

executor_config

num_hashes

= self.num_bands * self.minhashes_per_band

nemo_curator.stages.deduplication.fuzzy.workflow.FuzzyDeduplicationWorkflow._create_connected_components_pipeline() -> nemo_curator.pipeline.Pipeline

nemo_curator.stages.deduplication.fuzzy.workflow.FuzzyDeduplicationWorkflow._create_lsh_pipeline() -> nemo_curator.pipeline.Pipeline

nemo_curator.stages.deduplication.fuzzy.workflow.FuzzyDeduplicationWorkflow._create_minhash_pipeline(
    generate_input_filegroups: bool
) -> nemo_curator.pipeline.Pipeline

nemo_curator.stages.deduplication.fuzzy.workflow.FuzzyDeduplicationWorkflow._validate_initial_tasks(
    initial_tasks: list[nemo_curator.tasks.FileGroupTask] | None
) -> None

nemo_curator.stages.deduplication.fuzzy.workflow.FuzzyDeduplicationWorkflow._validate_inputs() -> None

nemo_curator.stages.deduplication.fuzzy.workflow.FuzzyDeduplicationWorkflow.run(
    initial_tasks: list[nemo_curator.tasks.FileGroupTask] | None = None,
    executor: nemo_curator.backends.ray_actor_pool.RayActorPoolExecutor | None = None
) -> nemo_curator.pipeline.workflow.WorkflowRunResult

Run the deduplication pipeline.

Parameters:

initial_tasks

list[FileGroupTask] | NoneDefaults to None

executor

RayActorPoolExecutor | NoneDefaults to None

Executor to use for the pipeline. Defaults to RayActorPoolExecutor().

nemo_curator.stages.deduplication.fuzzy.workflow.ID_GENERATOR_OUTPUT_FILENAME = 'fuzzy_id_generator.json'