Add ray data for audio modality

benchmarking/nightly-benchmark.yaml

@@ -580,15 +580,15 @@ entries:
       - metric: throughput_images_per_sec
         min_value: 3.0
 
-  - name: audio_fleurs
+  - name: audio_fleurs_xenna
     enabled: true
     script: audio_fleurs_benchmark.py
     args: >-
       --benchmark-results-path={session_entry_dir}
       --scratch-output-path={session_entry_dir}/scratch
       --model-name=nvidia/stt_hy_fastconformer_hybrid_large_pc
       --lang=hy_am
-      --split=dev
+      --split=train
       --wer-threshold=5.5
       --gpus=1
     ray:
@@ -600,6 +600,27 @@ entries:
         ping_on_failure:
           - U03C41SNADV  # Aaftab V
 
+  - name: audio_fleurs_raydata
+    enabled: true
+    script: audio_fleurs_benchmark.py
+    args: >-
+      --benchmark-results-path={session_entry_dir}
+      --scratch-output-path={session_entry_dir}/scratch
+      --model-name=nvidia/stt_hy_fastconformer_hybrid_large_pc
+      --lang=hy_am
+      --split=train
+      --wer-threshold=5.5
+      --gpus=1
+      --executor=ray_data
+    ray:
+      num_cpus: 64
+      num_gpus: 4
+      enable_object_spilling: false
+    sink_data:
+      - name: slack
+        ping_on_failure:
+          - U03C41SNADV  # Aaftab V
+
   - name: arxiv_e2e_pipeline_raydata
     enabled: true
     script: arxiv_e2e_pipeline_benchmark.py
@@ -888,3 +909,35 @@ entries:
         min_value: 1
       - metric: total_filtered_windows
         min_value: 1
+
+  - name: alm_pipeline_ray_data
+    enabled: true
+    script: alm_pipeline_benchmark.py
+    args: >-
+      --benchmark-results-path={session_entry_dir}
+      --input-manifest={curator_repo_dir}/tests/fixtures/audio/alm/sample_input.jsonl
+      --executor=ray_data
+      --target-window-duration=120.0
+      --tolerance=0.1
+      --min-sample-rate=16000
+      --min-bandwidth=8000
+      --min-speakers=2
+      --max-speakers=5
+      --overlap-percentage=50
+      --repeat-factor=2000
+    timeout_s: 600
+    sink_data:
+      - name: slack
+        ping_on_failure:
+          - U03C41SNADV  # Aaftab V
+    ray:
+      num_cpus: 8
+      num_gpus: 0
+      enable_object_spilling: false
+    requirements:
+      - metric: is_success
+        exact_value: true
+      - metric: total_builder_windows
+        min_value: 1
+      - metric: total_filtered_windows
+        min_value: 1

benchmarking/scripts/audio_fleurs_benchmark.py

@@ -15,10 +15,11 @@
 """Audio Fleurs benchmarking script.
 
 This script runs audio Fleurs benchmarks with comprehensive metrics collection
-using XennaExecutor and logs results to configured sinks.
+and logs results to configured sinks.
 """
 
 import argparse
+import traceback
 from pathlib import Path
 from typing import Any
 
@@ -59,6 +60,7 @@ def run_audio_fleurs_benchmark(  # noqa: PLR0913
         raise ValueError(msg)
 
     logger.info("Starting audio fleurs benchmark")
+    logger.info(f"Executor: {executor}")
     logger.info(f"Model: {model_name}")
     logger.info(f"Language: {lang}")
     logger.info(f"Split: {split}")
@@ -147,6 +149,10 @@ def main() -> int:
     try:
         result_dict.update(run_audio_fleurs_benchmark(**vars(args)))
         success_code = 0 if result_dict["metrics"]["is_success"] else 1
+    except Exception as e:
+        error_traceback = traceback.format_exc()
+        logger.error(f"Benchmark failed: {e}")
+        logger.debug(f"Full traceback:\n{error_traceback}")
     finally:
         write_benchmark_results(result_dict, args.benchmark_results_path)
     return success_code

nemo_curator/stages/audio/datasets/fleurs/create_initial_manifest.py

@@ -14,7 +14,9 @@
 
 import os
 from dataclasses import dataclass
+from typing import Any
 
+from nemo_curator.backends.experimental.utils import RayStageSpecKeys
 from nemo_curator.stages.audio.datasets.file_utils import download_file, extract_archive
 from nemo_curator.stages.base import ProcessingStage
 from nemo_curator.tasks import AudioBatch, _EmptyTask
@@ -138,6 +140,9 @@ def download_extract_files(self, dst_folder: str) -> None:
 
         extract_archive(f"{dst_folder}/{self.split}.tar.gz", str(dst_folder), force_extract=True)
 
+    def ray_stage_spec(self) -> dict[str, Any]:
+        return {RayStageSpecKeys.IS_FANOUT_STAGE: True}
+
     def process(self, _: _EmptyTask) -> list[AudioBatch]:
         self.download_extract_files(self.raw_data_dir)
         return self.process_transcript(os.path.join(self.raw_data_dir, self.split + ".tsv"))

tests/stages/audio/datasets/test_fleurs_create_initial_manifest.py

@@ -31,6 +31,15 @@ def _import_stage_module() -> tuple[Any, Any]:
     return CreateInitialManifestFleursStage, get_fleurs_url_list
 
 
+def test_ray_stage_spec(tmp_path: Path) -> None:
+    from nemo_curator.backends.experimental.utils import RayStageSpecKeys
+
+    stage_cls, _ = _import_stage_module()
+    stage = stage_cls(lang="hy_am", split="dev", raw_data_dir=str(tmp_path / "fleurs"))
+    spec = stage.ray_stage_spec()
+    assert spec[RayStageSpecKeys.IS_FANOUT_STAGE] is True
+
+
 def test_get_fleurs_url_list_builds_urls() -> None:
     _, get_fleurs_url_list = _import_stage_module()
     urls = get_fleurs_url_list("hy_am", "dev")