Dedupe pipeline on 10BT fineweb-edu sample (#2129)

## Description

Re: #2096

In this PR - "we use more data and more resources":
 * we add `fineweb_edu_small_10bt` dataset
 * request more resources - 1024 parallelism by default
 * guard `dupekit` to "protect" the Marin driver
 * cleanup the dedupe code slightly - still not clean enough
   * we will break out the dedupe utils out soon 
 * compute dedupe stats/counts
 * reuse more code between paragraph and doc exact modes

Author: ravwojdyla

experiments/dedup/dedup.py

@@ -50,6 +50,16 @@
     ),
 )
 
+fineweb_edu_small_10bt = ExecutorStep(
+    name="raw_fineweb_edu_small_10bt",
+    fn=download_hf,
+    config=DownloadConfig(
+        hf_dataset_id="HuggingFaceFW/fineweb-edu",
+        revision="3c452cb",
+        hf_urls_glob=["sample/10BT/*.parquet"],
+    ),
+)
+
 # N-gram configuration for train-test overlap detection
 DEFAULT_NGRAM_CONFIG = NGramConfig(
     ngram_length=[5, 10, 15],
@@ -61,14 +71,25 @@
 @ray.remote(runtime_env={"env_vars": {"JAX_PLATFORMS": "cpu", "PJRT_DEVICE": "cpu"}})
 def run_dedup(config: DedupeConfig) -> str:
     logger.info(f"Starting dedupe with config: {config}")
+
     dedupe(config)
+
     logger.info(f"Dedupe completed! Results written to {config.output_path}")
     return config.output_path
 
 
-def build_dedup_step(dataset: InputName) -> ExecutorStep:
+def build_dedup_step(dataset: InputName, max_parallelism: int) -> ExecutorStep:
+    """
+    Builds a deduplication step for the given dataset.
+
+    Args:
+        dataset: The input dataset to deduplicate.
+        max_parallelism: Maximum parallelism for Zephyr tasks.
+    """
+    input_path = dataset.cd("sample/10BT")
+
     config = DedupeConfig(
-        input_path=dataset.cd("sample/10BT"), attribute_name="is_duplicate", mode=DedupMode.EXACT_DOC_DEDUPLICATE
+        input_path=input_path, attribute_name="is_duplicate", mode=DedupMode.DEDUPLICATE, processes=max_parallelism
     )
 
     return ExecutorStep(
@@ -80,7 +101,11 @@ def build_dedup_step(dataset: InputName) -> ExecutorStep:
     )
 
 
-STEPS = [build_dedup_step(fineweb_edu_small_1), build_dedup_step(fineweb_edu_small_2)]
+STEPS = [
+    build_dedup_step(fineweb_edu_small_1, max_parallelism=7),
+    build_dedup_step(fineweb_edu_small_2, max_parallelism=7),
+    build_dedup_step(fineweb_edu_small_10bt, max_parallelism=1024),
+]
 
 if __name__ == "__main__":
     logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")