[error] too many open files in MinhashDedupBuckets process

The following error occurs while performing deduplication within the dataset.

OSError: [errno 23] Too many open files in system ~~

The work environment I run is a little complicated.

I'm using a slurm environment that uses docker images.

when I check **ulimit -n** in docker images, ulimit -n is 1,048,576. and ulimit -n in slurm node, ulimit -n is 500,000.
but signature files consist of 14 buckets, and each bucket has 14,000 files. Even if calculated overall, it is thought to be 14*14,000, which is less than 200,000 pieces.
So I don't think there will be a problem, but there is an error.

Do you have any idea why?

My machine has 1TB RAM, 256 CPU cores. my goal is to deduplicate 8TB text (14,000 jsonl files / each is 400MB and 100,000 lines)

---
Slurm script

sbatch run_dedup.sh

---
run_dedup.sh script
#SBATCH config
python fuzzy_dedup.py

---

fuzzy_dedup.py script

```
from datatrove.executor import LocalPipelineExecutor
from datatrove.pipeline.dedup import MinhashDedupSignature
from datatrove.pipeline.dedup.minhash import (
MinhashConfig,
MinhashDedupBuckets,
MinhashDedupCluster,
MinhashDedupFilter,
)
from datatrove.pipeline.readers import JsonlReader
from datatrove.pipeline.tokens import TokensCounter
from datatrove.pipeline.writers.jsonl import JsonlWriter

# you can also change ngrams or the number of buckets and their size here
minhash_config = MinhashConfig(use_64bit_hashes=True) # better precision -> fewer false positives (collisions)
MINHASH_BASE_PATH = "minhash/output"
LOGS_FOLDER = "minhash/log"
LOCAL_LOGS_FOLDER = "my_local_folder_for_slurm_logs/"

TOTAL_TASKS = 14000
NUM_WORKERS = 128

# this is the original data that we want to deduplicate
INPUT_READER = JsonlReader(
data_folder="data",
recursive=True ## I have some subfolders.
)

# stage 1 computes minhash signatures for each task (each task gets a set of files)
stage1 = LocalPipelineExecutor(
pipeline=[
INPUT_READER,
MinhashDedupSignature(
output_folder=f"{MINHASH_BASE_PATH}/signatures",
config=minhash_config
),
],
tasks=TOTAL_TASKS,
workers=NUM_WORKERS,
logging_dir=f"{LOGS_FOLDER}/signatures",
)

# stage 2 finds matches between signatures in each bucket
stage2 = LocalPipelineExecutor(
pipeline=[
MinhashDedupBuckets(
input_folder=f"{MINHASH_BASE_PATH}/signatures",
output_folder=f"{MINHASH_BASE_PATH}/buckets",
config=minhash_config,
),
],
tasks=minhash_config.num_buckets,
workers=NUM_WORKERS,
logging_dir=f"{LOGS_FOLDER}/buckets",
depends=stage1,
)

# stage 3 creates clusters of duplicates using the results from all buckets
stage3 = LocalPipelineExecutor(
pipeline=[
MinhashDedupCluster(
input_folder=f"{MINHASH_BASE_PATH}/buckets",
output_folder=f"{MINHASH_BASE_PATH}/remove_ids",
config=minhash_config,
),
],
tasks=1,
workers=NUM_WORKERS,
logging_dir=f"{LOGS_FOLDER}/clusters",
depends=stage2,
)

stage4 = LocalPipelineExecutor(
pipeline=[
INPUT_READER,
TokensCounter("Llama-3-8B/tokenizer.json"), # nice way to see how many tokens we had before and after deduplication
MinhashDedupFilter(
input_folder=f"{MINHASH_BASE_PATH}/remove_ids",
exclusion_writer=JsonlWriter(f"{MINHASH_BASE_PATH}/removed"),
),
JsonlWriter(output_folder=f"{MINHASH_BASE_PATH}/deduplicated_output"),
],
tasks=TOTAL_TASKS,
workers=NUM_WORKERS,
logging_dir=f"{LOGS_FOLDER}/filter",
depends=stage3,
)

stage4.run()
```