FuzzyDeduplicationWorkflow crashes due to multiple bugs

[jmcmanus15]

See Issue #1216, where I first reported the crashes.

To reproduce:

1. Run the nvcr.io/nvidia/nemo-curator:25.09 docker image
2. Inside the container, execute the code example in the documentation: https://docs.nvidia.com/nemo/curator/latest/curate-text/process-data/deduplication/index.html. You have to start the ray client before running the workflow (which isn't described in the docs but is consistent with the semantic dedup tutorial).

That's it. I can't share the data I'm using because they're proprietary, but you can choose any data you like. I stepped through the code in the debugger and investigated debugging prints. The bugs are unrelated to my data.

See Issue #1216 for details about all the things that go wrong when you attempt the above. It fails immediately. If you circumvent one problem, you immediately find another. I can't even get the first stage (MinHash) to succeed, much less the full workflow.

I'd consider this issue resolved if someone can produce a single example that actually works with FuzzyDeduplicationWorkflow. That's all I want at this point.

[jmcmanus15]

Perhaps the issue is that the FuzzyDeduplicationWorkflow uses the "experimental" Ray backend, and this isn't ready for primetime? But how can I use the XennaExecutor instead? Would I need to implement the whole workflow step-by-step perhaps? And even then, some of the issues I found appear to be problems in MinHashStage; is that ray-specific?

[jmcmanus15]

See the resolution in Issue #1216.

The crashes were due to my own user error:

1. The input JSONL files must not contain any example documents shorter than the shingle length (at least, must not contain any empty docs)
2. This workflow requires a BIG machine. I had to use 8x B200s to get what I wanted, on a corpus of only 19 GB. The crashes I was getting were merely OOM errors.

I'm marking this issue as resolved.

All of the above said, I'd suggest some small updates to the documentation, to make these points a little clearer. Ideally, the preconditions on the input would be spelled out, and some rough guidance on how much VRAM is needed would be helpful. Or, at the very least, some warnings in the docs about what errors to expect if you don't satisfy the preconditions. Adding input validation to the pipeline/workflow is probably too much to ask, so I won't try :-)

[ayushdg]

Thanks for testing it out and sharing your findings. Some of them are not expected and do need to be investigated further and I'll open up some issues to follow up.

Things to improve:

• Documentation and examples
• When document length is lesser than shingle length the error message is very confusing. We need to do a better job raising more informative errors.
• I've generally had success working on a corpus of 19GB and deduplicating it with 2 A100 GPUs using the default parameters. The fact that you needed to go so much bigger in compute is a bit surprising. In most of my testing a 19GB corpus (~4million documents) should comforatably fit and I've been able to scale upto 8TB (red pajama v2 data) on an 8X B200 node. @jmcmanus15 Is it possible to share the row count of the dataset and the parameters you used for the successful run on the 8x B200.

Somewhat expected issue:

• Scaling up to 9000 hashes (refined web): Irrespective of overall compute, each batch of input data needs to generate and hold on to 9000 hashes (450*20) before writing out to parquet and significantly increases the memory overhead of each batch, so overall cluster size doesn't help here and having fewer rows per batch (smaller blocksize) or a GPU with larger memory (B200 is the largest unfortunately).
  For that scale rather than computing all 9000 hashes at once it might be more memory efficient if we generated only 450 hashes at a time (1 band) computed the band ID and just kept that in memory instead. I'll open an issue to showcase an example running this with those parameters.

[jmcmanus15]

Hey, @ayushdg

Just to be clear, I never tested the default params (20 bands and 13 hashes per band) on the smaller machines. At that point, I didn't realize I was hitting OOM errors. I started to suspect it, then ran on the big machines (8x B200s). When even that crashed, that's when I realized 9000 hashes per document must be way too much for this pipeline. I didn't drop down to 20 bands until I was already on the big machine.

There are 13 million documents in my dataset. These params crashed even the big boy:

• input_blocksize="256MB"
• num_bands=450
• minhashes_per_band=20
• bands_per_iteration=5

Dropping down to the defaults made the run succeed on the big machine:

• input_blocksize="256MB"
• num_bands=20
• minhashes_per_band=13
• bands_per_iteration=5

With these settings, the whole pipeline ran in less than 272 seconds, which is just absurdly fast.

Eventually, I started thinking more clearly about how all these parameters influence memory usage. So today I found a way to make 9000 hashes work on the big machine. These params work:

• input_blocksize="32MB"
• num_bands=450
• minhashes_per_band=20
• bands_per_iteration=1

With these settings, the pipeline runs in 16,290 seconds! A lot slower. I checked the GPU memory usage often throughout the run. VRAM usage at its highest was around 1.3 TB, close to the 1.46 TB capacity of 8 B200s.

Actually, overall the pipeline is still fast with these settings. It's just the LSH stage that takes forever, because there are now so many bands, and each iteration only processes one at a time. Painful.

With the benefit of hindsight, I'd say a good compromise to achieve the desired S-curve would be something like:

• input_blocksize="32MB"
• num_bands=225
• minhashes_per_band=15
• bands_per_iteration=2

That should be about 4 times faster, I'd guess, while generating "superior" results over the default band settings.