Implementation:NVIDIA NeMo Curator RayDataExecutor

Knowledge Sources	NVIDIA NeMo Curator
Domains	Backend Architecture, Ray Integration, Ray Data, Pipeline Execution
Last Updated	2026-02-14 00:00 GMT

Overview

Implements an experimental pipeline executor that uses Ray Data datasets as the data transport, converting tasks to a Ray Data dataset and applying each stage as a map_batches transformation.

Description

RayDataExecutor extends BaseExecutor to provide a pipeline execution engine built on Ray Data. It logs a warning at construction that this executor is experimental and may not work as expected.

The execute() method follows this sequence:

1. Initialize Ray -- Calls ray.init() with ignore_reinit_error=True and sets RAY_EXPERIMENTAL_NOSET_CUDA_VISIBLE_DEVICES to empty string (overriding any Xenna-set value).
2. Convert tasks to dataset -- Converts the initial task list (or a single EmptyTask if none provided) to a Ray Data dataset via ray.data.from_items(), with one block per task.
3. Setup on node -- Calls execute_setup_on_node() for all stages across all cluster nodes.
4. Process stages -- Iterates through each stage, creating a RayDataStageAdapter and calling process_dataset() to apply the stage as a Ray Data transformation. Each stage's CPU and GPU resource requirements are logged.
5. Materialize results -- After all stages, calls take_all() on the final dataset and extracts Task objects from the "item" column.
6. Cleanup -- Calls ray.shutdown() in a finally block to clean up resources.

Error handling wraps the entire execution in a try/except/else/finally block, logging errors and ensuring Ray is always shut down.

Usage

Use RayDataExecutor when you want to leverage Ray Data's built-in streaming execution, automatic memory management, and fault tolerance. Note that this executor is experimental and does not currently support RAFT or shuffle stages that are available in the ActorPool executor.

Code Reference

Source Location

• Repository: NeMo-Curator
• File: nemo_curator/backends/experimental/ray_data/executor.py
• Lines: 1-131

Signature

class RayDataExecutor(BaseExecutor):
    def __init__(self, config: dict[str, Any] | None = None, ignore_head_node: bool = False): ...
    def execute(self, stages: list[ProcessingStage], initial_tasks: list[Task] | None = None) -> list[Task]: ...

Import

from nemo_curator.backends.experimental.ray_data.executor import RayDataExecutor

I/O Contract

Inputs

Name	Type	Required	Description
config	dict[str, Any] or None	No	Optional configuration dictionary for the executor
ignore_head_node	bool	No	Whether to ignore the head node for scheduling (default False)

execute() Inputs

Name	Type	Required	Description
stages	list[ProcessingStage]	Yes	List of processing stages to execute in order
initial_tasks	list[Task] or None	No	Initial tasks to process; if None, a single EmptyTask is used

Outputs

Name	Type	Description
execute()	list[Task]	Final processed tasks after all stages complete; empty list if no stages provided

Internal Methods

Method	Description
_tasks_to_dataset(tasks)	Converts a list of Task objects to a Ray Data dataset using ray.data.from_items() with one block per task
_dataset_to_tasks(dataset)	Converts a Ray Data dataset back to a list of Task objects by calling take_all() and extracting the "item" column

Usage Examples

Running a Pipeline with RayDataExecutor

from nemo_curator.backends.experimental.ray_data.executor import RayDataExecutor

# Create executor
executor = RayDataExecutor(config={"some_option": "value"}, ignore_head_node=True)

# Execute a pipeline of stages
results = executor.execute(stages=[stage1, stage2, stage3], initial_tasks=my_tasks)

Using with Default Configuration

from nemo_curator.backends.experimental.ray_data.executor import RayDataExecutor

executor = RayDataExecutor()
results = executor.execute(stages=my_stages)
# results is a list[Task] with the output from the last stage

Related Pages

• Environment:NVIDIA_NeMo_Curator_Python_Linux_Base
• NVIDIA_NeMo_Curator_Backend_Base_Classes -- Parent base classes
• NVIDIA_NeMo_Curator_RayDataStageAdapter -- Adapter used to transform stages into Ray Data operations
• NVIDIA_NeMo_Curator_XennaExecutor -- Production executor alternative