Add snapshot-ready MinerU-HTML extraction with managed Dynamo

.github/workflows/cicd-main.yml

@@ -91,6 +91,14 @@ jobs:
         run: |
           sudo apt-get update
           sudo apt-get install -y --no-install-recommends ffmpeg
+      # The MinerU-HTML Markdown converter (webpage_converter) imports cairosvg on
+      # its first conversion, which dlopens libcairo.so.2. Install it explicitly
+      # rather than relying on it arriving transitively via the runner image.
+      - name: Install system dependencies for MinerU-HTML tests
+        if: matrix.folder == 'stages-text'
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y --no-install-recommends libcairo2
       - name: Install uv
         uses: astral-sh/setup-uv@d0cc045d04ccac9d8b7881df0226f9e82c39688e # v6
         with:

.github/workflows/install-test.yml

@@ -99,7 +99,7 @@ jobs:
         run: |
           set -o pipefail
           INSTALLER="${{ matrix.installer }}"
-          CPU_EXTRAS=("audio_cpu" "image_cpu" "sdg_cpu" "text_cpu" "video_cpu")
+          CPU_EXTRAS=("audio_cpu" "image_cpu" "mineru_html" "sdg_cpu" "text_cpu" "video_cpu")
           RESULTS=""
           FAILED=0
 

benchmarking/mineru-html-benchmark.yaml

@@ -0,0 +1,57 @@
+# One native Common Crawl shard per Slurm GPU node. Curator assigns complete
+# WARC source tasks deterministically from the Slurm-array environment.
+paths:
+  - name: results_path
+    host_path: "${MINERU_RESULTS_ROOT}"
+
+default_timeout_s: 7200
+max_timeout_s: 14340
+cleanup_timeout_s: 60
+min_timeout_s: 600
+gpu_stats_recorder:
+  interval_s: 1.0
+
+entries:
+  - name: mineru_cc_work_unit
+    script: mineru_html_benchmark.py
+    args: >-
+      --benchmark-results-path={session_entry_dir}
+      --snapshot=${MINERU_SNAPSHOT}
+      --warc-manifest=${MINERU_WARC_MANIFEST}
+      --output-path=${MINERU_OUTPUT_PATH}
+      --checkpoint-path=${MINERU_CHECKPOINT_PATH}
+      --download-dir=${MINERU_DOWNLOAD_DIR}
+      --download-workers=${MINERU_DOWNLOAD_WORKERS}
+      --warc-records-per-batch=${MINERU_WARC_RECORDS_PER_BATCH}
+      --cc-transport=${MINERU_CC_TRANSPORT}
+      --cc-s3-bucket=${MINERU_CC_S3_BUCKET}
+      --cc-s3-key-prefix=${MINERU_CC_S3_KEY_PREFIX}
+      --cc-s3-endpoint-url=${MINERU_CC_S3_ENDPOINT_URL}
+      --cc-s5cmd-concurrency=${MINERU_CC_S5CMD_CONCURRENCY}
+      --cc-s5cmd-part-size-mb=${MINERU_CC_S5CMD_PART_SIZE_MB}
+      --html-field=${MINERU_HTML_FIELD}
+      --html-compression=${MINERU_HTML_COMPRESSION}
+      --drop-html-field
+      --url-field=${MINERU_URL_FIELD}
+      --cache-dir=${MINERU_MODEL_CACHE}
+      --server-mode=managed
+      --reuse-driver-environment
+      --num-replicas=8
+      --kv-cache-dtype=fp8
+      --speculative-tokens=16
+      --structured-outputs=per_request
+      --cudagraph-mode=FULL_AND_PIECEWISE
+      --server-concurrency=256
+      --inference-workers=${MINERU_INFERENCE_WORKERS}
+      --simplify-workers=${MINERU_SIMPLIFY_WORKERS}
+      --extract-workers=${MINERU_EXTRACT_WORKERS}
+      --executor=ray_data
+    timeout_s: 12600
+    object_store_size: ${MINERU_OBJECT_STORE_SIZE}
+    ray:
+      num_gpus: 8
+    requirements:
+      - metric: is_success
+        exact_value: true
+      - metric: verification_passed
+        exact_value: true

benchmarking/mineru-html-snapshot-verify.yaml

@@ -0,0 +1,30 @@
+# CPU-only dependent job. It opens every published Parquet footer and only then
+# writes the snapshot-level success record.
+paths:
+  - name: results_path
+    host_path: "${MINERU_RESULTS_ROOT}"
+
+gpu_stats_recorder:
+  interval_s: 0
+
+entries:
+  - name: mineru_cc_snapshot_verify
+    script: mineru_html_benchmark.py
+    args: >-
+      --benchmark-results-path={session_entry_dir}
+      --verify-snapshot
+      --snapshot=${MINERU_SNAPSHOT}
+      --warc-manifest=${MINERU_WARC_MANIFEST}
+      --output-path=${MINERU_OUTPUT_PATH}
+      --checkpoint-path=${MINERU_CHECKPOINT_PATH}
+      --snapshot-success-path=${MINERU_SNAPSHOT_SUCCESS_PATH}
+    timeout_s: 1800
+    object_store_size: 4294967296
+    ray:
+      num_gpus: 0
+      num_cpus: 4
+    requirements:
+      - metric: verification_passed
+        exact_value: true
+      - metric: num_output_files
+        min_value: 1

benchmarking/mineru-html-snapshot.md

@@ -0,0 +1,124 @@
+# MinerU-HTML: one Common Crawl snapshot
+
+This path streams a main Common Crawl snapshot through native NeMo Curator
+stages. It does not materialize an intermediate HTML dataset or maintain a
+second work-unit manifest.
+
+## Data and ownership model
+
+`CommonCrawlWARCManifestSourceStage` emits one deterministic source task per
+WARC from the frozen snapshot manifest; it performs no URL discovery.
+Curator's Slurm-array source filter assigns each complete WARC to exactly one
+logical shard. A fused `CommonCrawlWARCDownloadAndReadStage` then:
+
+1. downloads the whole WARC to allocation-local storage;
+2. reads the WARC on that same Ray worker;
+3. compresses each response body as an independent Zstandard frame;
+4. emits bounded 1,024-record batches so simplify and conversion use all of the
+   configured CPU workers;
+5. removes the local WARC before the task leaves the worker.
+
+The fused stage is deliberate. Separate download and iteration stages may land
+on different Ray nodes, which is invalid when the download directory is local
+RAID. The compressed HTML survives only long enough for MinerU fallback and is
+dropped before the output Parquet is written.
+
+Every array element has one four-hour, eight-GPU node and is launched with
+`SlurmRayClient`. `Pipeline.run(checkpoint_path=...)` records completed source
+WARCs, so a retry re-downloads only unfinished sources. The checkpoint and
+output paths must be shared; the WARC and Ray temp paths must be local to the
+allocation. The Slurm launcher prefers its job-specific directory under
+`/raid/scratch/$USER` and falls back to `SLURM_TMPDIR` or `/tmp` only when local
+RAID is unavailable. Ray uses a separate short `r<jobid>-<array-index>` temp
+root on the same filesystem so its generated UNIX socket stays below Linux's
+107-byte path limit.
+
+The 128-CPU node baseline uses 2 whole-WARC download, 32 simplify, 32
+inference-client, and 32 extraction actors. Each downloader accepts only one
+in-flight WARC, preventing multi-GiB fan-out results from crowding CPU work out
+of Ray's execution budget. This leaves 30 CPUs unreserved for vLLM, Dynamo,
+Ray, writers, and the operating system. Its 256 GiB Ray object store leaves a
+128 GiB Ray Data execution budget for overlapping compressed, simplified, and
+inferred batches on the measured 1.5 TiB H100 node. Override these independently with
+`MINERU_DOWNLOAD_WORKERS`, `MINERU_SIMPLIFY_WORKERS`,
+`MINERU_INFERENCE_WORKERS`, and `MINERU_EXTRACT_WORKERS` after measuring a
+representative canary on the target node type; override the object store with
+`MINERU_OBJECT_STORE_SIZE` only when shared-memory capacity has been verified.
+
+## Serving baseline
+
+The production baseline is vLLM 0.26, Dynamo 1.4,
+`FULL_AND_PIECEWISE` CUDA graphs, FP8 KV cache, per-request structured output,
+and suffix speculation with 16 draft tokens through ArcticInference. Native
+async scheduling remains disabled because this suffix path does not support it.
+The Curator defaults are used for model context (`32768`) and per-element DOM
+cutoff (`500`).
+
+## Submit
+
+Install the locked environment once at a shared path:
+
+```bash
+uv sync --frozen --extra mineru_html_inference --extra text_cpu
+```
+
+Then export the shared locations and submit:
+
+```bash
+export CURATOR_DIR=/shared/checkouts/Curator
+export MINERU_RESULTS_ROOT=/shared/cc/CC-MAIN-2025-26/run-results
+export MINERU_OUTPUT_PATH=/shared/cc/CC-MAIN-2025-26/mineru-output
+export MINERU_CHECKPOINT_PATH=/shared/cc/CC-MAIN-2025-26/mineru-checkpoint
+export MINERU_SNAPSHOT_SUCCESS_PATH=/shared/cc/CC-MAIN-2025-26/SNAPSHOT_SUCCESS.json
+export MINERU_MODEL_CACHE=/shared/huggingface/hub
+export MINERU_SNAPSHOT=2025-26
+export CURATOR_DYNAMO_BIN_DIR=/shared/bin
+export MINERU_SLURM_ACCOUNT=<slurm-account>
+
+MINERU_TOTAL_SHARDS=1400 \
+MINERU_MAX_GPU_NODES=32 \
+benchmarking/slurm/submit_mineru_cc_snapshot.sh
+```
+
+`1400` is a conservative initial split based on the measured 1M replay rate;
+WARC record counts vary, so run a small canary cohort and adjust the shard count
+before the full launch. A logical shard count is immutable once its checkpoint
+path exists. Use a new checkpoint path if it changes.
+
+The production transport is whole-object multipart download from the internal
+PDX mirror through `s5cmd`; it does not contact `data.commoncrawl.org`.
+`pdx-commoncrawl` in `~/.config/datamover/storage_locations` supplies credentials
+and the `https://pdx.s8k.io` endpoint. Official manifest entries map from
+`crawl-data/CC-MAIN-...` to bucket/key `s3://crawl-data/CC-MAIN-...`. Each of the
+2 download actors transfers one WARC at a time with up to 8 concurrent 256 MiB
+parts. The Slurm preflight resolves the Data Mover location without logging its
+secret and verifies one exact manifest object before starting Ray or the model.
+
+The storage measurements show that PDX is request-rate-bound rather than
+bandwidth-bound. Whole WARC objects keep requests large and the default transfer
+shape stays far below the measured per-key and store-wide request ceilings.
+Override `MINERU_DM_STORAGE_LOCATION`, `MINERU_CC_S3_ENDPOINT_URL`,
+`MINERU_CC_S3_BUCKET`, `MINERU_CC_S3_KEY_PREFIX`,
+`MINERU_CC_S5CMD_CONCURRENCY`, or `MINERU_CC_S5CMD_PART_SIZE_MB` only for a
+different verified mirror layout.
+
+## Verification and retries
+
+Each successful shard writes Curator's native completion manifest. The
+dependent CPU job verifies:
+
+- every logical shard has a completion manifest;
+- output Parquet count is at least the snapshot WARC count (each WARC fans out
+  into deterministic bounded chunks);
+- every Parquet footer is readable and contains `url`, `text`, and
+  `_mineru_status`;
+- quality rates pass on an evenly distributed 1,024-file sample.
+
+Only then is `SNAPSHOT_SUCCESS.json` written. This proves operational
+completeness and checks gross quality drift; it does not replace the labelled
+100k F1 canary when model, prompt, parser, or serving versions change.
+
+If a shard fails, its completion manifest is absent. Resubmit the missing
+logical shard indices against the same checkpoint and output paths, then submit
+a new verifier dependency. Do not combine an external work-unit manifest with
+native Slurm-array sharding.

benchmarking/runner/ray_cluster.py

@@ -24,7 +24,7 @@
 from loguru import logger
 from runner.utils import get_shm_usage
 
-from nemo_curator.core.client import RayClient
+from nemo_curator.core.client import RayClient, SlurmRayClient
 from nemo_curator.core.utils import check_ray_responsive
 
 ray_client_start_timeout_s = 30
@@ -53,8 +53,10 @@ def setup_ray_cluster_and_env(  # noqa: PLR0913
     include_dashboard: bool = True,
 ) -> tuple[RayClient, Path]:
     """Setup a Ray cluster and set the RAY_ADDRESS environment variable and return the Ray client and temp dir."""
-    # Create a short temp dir to avoid Unix socket path length limits
-    short_temp_path = Path(f"/tmp/ray_{uuid.uuid4().hex[:8]}")  # noqa: S108
+    # Prefer allocation-local storage under Slurm. Keep the random suffix so
+    # retries and multiple benchmark entries never share Ray sockets/state.
+    ray_temp_root = Path(os.environ.get("RAY_TMPDIR", "/tmp"))  # noqa: S108
+    short_temp_path = ray_temp_root / f"ray_{uuid.uuid4().hex[:8]}"
     short_temp_path.mkdir(parents=True, exist_ok=True)
 
     # Capture stdout/stderr to a file if provided, otherwise suppress it
@@ -81,7 +83,7 @@ def setup_ray_cluster_and_env(  # noqa: PLR0913
             ray_stdouterr_capture_file = f"{ray_log_path!s}-{retries + 1}"
 
         # Create and start the Ray client
-        client = RayClient(
+        client = SlurmRayClient(
             ray_temp_dir=str(short_temp_path),
             include_dashboard=include_dashboard,
             num_gpus=num_gpus,
@@ -114,7 +116,8 @@ def setup_ray_cluster_and_env(  # noqa: PLR0913
         msg = f"Failed to start Ray cluster after {max_retries} attempts"
         raise RuntimeError(msg)
 
-    logger.info(f"RayClient started successfully: pid={client.ray_process.pid}, port={client.ray_port}")
+    pid = client.ray_process.pid if client.ray_process else None
+    logger.info(f"SlurmRayClient started successfully: pid={pid}, port={client.ray_port}")
     return client, short_temp_path