Add ray data for audio modality

[oyilmaz-nvidia]

Description

This PR adds a few changes to test and benchmark Ray Data for audio workflows.

Fanouts

FLEURS is the only audio dataset with an initial manifest creation stage in this PR. The fanout is needed because CreateInitialManifestFleursStage takes a single input task and produces many AudioBatch tasks (one per transcript line). The other audio stages added in this PR (GetAudioDurationStage, ALMManifestWriterStage, InferenceAsrNemoStage) are 1-to-1 transforms, so they don't need fanout.

ALMManifestReaderStage also has IS_FANOUT_STAGE: True (it reads a JSONL and fans out one task per entry), but that's a pre-existing stage that just got ray_stage_spec() added.

Should You Add More Fanouts?

Not right now — the fanout is tied to a specific behavioral pattern (one-input → many-outputs), not to the audio modality itself. You'd only add fanouts for new audio stages that:

Checklist

• I am familiar with the Contributing Guide.
• New or Existing tests cover these changes.
• The documentation is up to date with these changes.

[copy-pr-bot]

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

[oyilmaz-nvidia]

/ok to test a90a3af

[oyilmaz-nvidia]

/ok to test 235ef99

[praateekmahajan]

Thanks @oyilmaz-nvidia

In the PR description can you explain the intent to change the stages to actors, and similarly which ones you converted to fanout and why, vs which ones you didn't and why?

[mohammadaaftabv]

To run alm benchmark on large scale real world data, below are the following steps:

1. Get data here. Size is 1.17 GB.
2. Update --input-manifest= param with path of said download.
3. set --repeat-factor=1 in this command.
4. set timeout_s: 6000 or max permissible.

Same instructions for both xenna and ray.
Based on completed time for repeat-factor=1; consider increasing repeat factor to any number between 2-10.

[mohammadaaftabv]

For large scale benchmarks, here set --split=train. You should run on ~9x more data.

[oyilmaz-nvidia]

/ok to test da6bf1e

[greptile-apps]

Syntax error: method definition floating between module-level imports

Lines 19–22 contain an indented def ray_stage_spec(self) body placed directly between top-level import statements. Python will raise an IndentationError the moment this module is collected, making the entire module unimportable. This is the complete implementation of the method (including the deferred RayStageSpecKeys import) that should have been placed inside the class body — it was apparently mis-applied here by mistake.

These four lines must be removed from the module scope entirely. The correct implementation already exists at line 146, but it is incomplete (see the companion comment below).

Suggested change

	def ray_stage_spec(self) -> dict[str, Any]:
	from nemo_curator.backends.experimental.utils import RayStageSpecKeys
	return {RayStageSpecKeys.IS_FANOUT_STAGE: True}
	from nemo_curator.stages.audio.datasets.file_utils import download_file, extract_archive
	from nemo_curator.stages.audio.datasets.file_utils import download_file, extract_archive

[oyilmaz-nvidia]

/ok to test 3b4cb6c

[greptile-apps]

No timeout_s for train-split benchmarks

Both audio_fleurs_raydata (new) and audio_fleurs_xenna (changed from dev → train) now process the full Armenian FLEURS training split, which is considerably larger than dev. Without a timeout_s, a slow download, a hung Ray worker, or unexpectedly slow ASR inference could cause these entries to block the entire nightly run indefinitely.

The alm_pipeline_ray_data entry added in this same PR does include timeout_s: 600 for reference. Consider adding a generous but bounded timeout to both audio-fleurs entries, e.g.:

  - name: audio_fleurs_raydata
    enabled: true
    timeout_s: 7200   # 2 h ceiling for train-split download + inference
    script: audio_fleurs_benchmark.py
    ...

and similarly for audio_fleurs_xenna.

[mohammadaaftabv]

Looks Good to me.

[praateekmahajan]

PR LGTM. I do think we should run with a larger dataset than this since a 2-3 minute tests won't tell us much about the autoscaling behaviors of the frameworks or help us catch regressions

[sarahyurick]

/ok to test bd25cbc

[greptile-apps]

Module-level Ray import still present

RayStageSpecKeys is imported at module level (line 19), and nemo_curator/backends/experimental/utils.py has import ray at its top level (line 19 of that file). This means every import of CreateInitialManifestFleursStage now transitively requires Ray to be installed — even in Xenna-only environments.

Concretely, test_get_fleurs_url_list_builds_urls and test_process_transcript_parses_tsv both call _import_stage_module(), which now pulls in ray and will raise an ImportError in environments without Ray installed. Since ray_stage_spec() is only ever called by the Ray Data backend, the import should be deferred to inside that method:

def ray_stage_spec(self) -> dict[str, Any]:
    from nemo_curator.backends.experimental.utils import RayStageSpecKeys
    return {RayStageSpecKeys.IS_FANOUT_STAGE: True}

Then remove the module-level from nemo_curator.backends.experimental.utils import RayStageSpecKeys on line 19.

[greptile-apps]

Missing requirements for audio_fleurs_raydata and audio_fleurs_xenna

Neither audio_fleurs_raydata nor audio_fleurs_xenna (both processing the full train split now) define a requirements block, while the other new entry in this PR — alm_pipeline_ray_data — does:

requirements:
  - metric: is_success
    exact_value: true
  - metric: total_builder_windows
    min_value: 1
  - metric: total_filtered_windows
    min_value: 1

Without requirements, the nightly framework cannot distinguish between a run that crashes (and writes is_success: false) and a run that silently produces zero output records (all audio filtered by WER threshold). The benchmarking framework will mark these entries as "passed" as long as the script exits without an unhandled exception.

Consider adding at minimum an is_success requirement for both entries, and ideally a min_value guard on a count metric (e.g., number of audio samples that passed the WER filter) to catch regressions in result quality.