Removal logic for Exact / Fuzzy Dedup

[praateekmahajan]

Description

Removal Description

1. We implement ~left-anti join using a broadcast merge. This allows us to scale even when right is greater than memory per node.
2. We observe that the performance varies as num partitions (or partition sizes vary).
   • The fastest being where right is exactly one partition, in which case we broadcast all of right to all the workers, and perform a merge (resulting in least transfer).
3. One downside of this approach is when right is bigger than system memory per node, then the merge operation will have to spill, and that won't be possible, hence resulting in a stuck state where dask pauses workers once it reaches 80% capacity in which case user will have to implement a different merge logic.

Class Description

1. Implements an abstract class called Deduplicator (name suggestions welcome)
   1. Lives in nemo_curator._deduplicator.py
   7. Implements identify / remove / call
   8. call now accepts a boolean called perform_removal (alternative API's also accepted)
2. Fuzzy / Exact Duplicates extend this class and the __call__ is now renamed as identify since the __call__ is implemented in base class

Usage

dataset = DocumentDataset.read_parquet(...)
exact_dedup = ExactDeduplicator(..)
duplicates = exact_dedup(dataset) # or exact_dedup.identify(exact_dedup)

exact_dedup.remove(dataset)

Checklist

• I am familiar with the Contributing Guide.
• New or Existing tests cover these changes.
• The documentation is up to date with these changes.

[ryantwolf]

Couple of comments about the organization. Can you make sure you update the API docs and deduplication sections in the user guide to mention all this?

One downside of this approach is when right is bigger than system memory per node, then the merge operation will have to spill, and that won't be possible, hence resulting in a stuck state where dask pauses workers once it reaches 80% capacity in which case user will have to implement a different merge logic.

Make sure to mention this in the user guide and describe roughly when "right is bigger than system memory per node" in layman's terms. Is this something we have/will encounter? Wondering if globally deduping 240 TB would trigger this.

[ryantwolf]

Can this be moved under nemo_curator/modules/deduplicator.py? I think the function is something we want users to be able to access.

[ryantwolf]

Not sure how I feel about the abstraction. I have been wanting something like this, but I worry this is not as generalizable as I'd want it to be. For example, can semantic dedupe use this? I don't believe it can since the duplicates aren't all grouped like this. Imo, if the deduplication abstraction doesn't work for all our deduplication methods I don't want to have it so we don't confuse our users. We can always refactor out the logic into a base class if we find a general solution.

[sarahyurick]

Maybe call it

Suggested change

	class Deduplicator(ABC):
	class DuplicateRemover:

or DuplicatesRemover instead?

[sarahyurick]

Possible example usage:

remover = DuplicatesRemover(...)

exact_dupes = ExactDuplicates(...).identify_duplicates(...)
deduped_data = remover.remove_duplicates(exact_dupes)

fuzzy_dupes = FuzzyDuplicates(...).identify_duplicates(...)
deduped_data = remover.remove_duplicates(fuzzy_dupes)

# Could it be possible to call both simultaneously?
# deduped_data = remover.remove_duplicates(exact_dupes, fuzzy_dupes)
# deduped_data = remover.remove_duplicates([exact_dupes, fuzzy_dupes])
# or similar...

?

[ryantwolf]

I think this function could just be a helper function that's defined here that exact and fuzzy dedup import and use instead of the ABC.

[sarahyurick]

Maybe call it

Suggested change

	class Deduplicator(ABC):
	class DuplicateRemover:

or DuplicatesRemover instead?

[sarahyurick]

Suggested change

	class ExactDuplicates(Deduplicator):
	class ExactDuplicates:

?

[sarahyurick]

Suggested change

	class FuzzyDuplicates(Deduplicator):
	class FuzzyDuplicates:

?

[sarahyurick]

Possible example usage:

remover = DuplicatesRemover(...)

exact_dupes = ExactDuplicates(...).identify_duplicates(...)
deduped_data = remover.remove_duplicates(exact_dupes)

fuzzy_dupes = FuzzyDuplicates(...).identify_duplicates(...)
deduped_data = remover.remove_duplicates(fuzzy_dupes)

# Could it be possible to call both simultaneously?
# deduped_data = remover.remove_duplicates(exact_dupes, fuzzy_dupes)
# deduped_data = remover.remove_duplicates([exact_dupes, fuzzy_dupes])
# or similar...

?

[praateekmahajan]

@sarahyurick / @ryantwolf given both of you had thoughts about it not being an abstract class, I took your suggestions and started this #509
@ryantwolf I'll add the docs once we are clear on what the API is, so that I don't have to change them repeatedly.

Regarding your question on if we've run into a case where it's greater than system (host) memory yes.
In one of our experiments, a 64TB dataset produced ~300gb of duplicates, and our host memory was 250gb per worker, which didn't work out for us, as CUDF during merge was spilling to host memory, and running OOM.
In that case, we'd likely recommend the user, to implement something custom where they do it at a partition level (where each partition typically is a year of CC or something similar)

It's still an open question on why dask / cudf needed to have the whole ~300gb in memory, but for now at that scale this won't work. It works wherever the duplicates size << host memory which can vary for different systems.