Implementation:NVIDIA NeMo Curator XennaExecutor

Knowledge Sources	NVIDIA NeMo Curator
Domains	Backend Architecture, Cosmos Xenna, Pipeline Execution, Distributed Computing
Last Updated	2026-02-14 00:00 GMT

Overview

Implements the default production pipeline executor that runs NeMo Curator stages through the Cosmos-Xenna distributed execution engine, supporting both streaming and batch execution modes.

Description

XennaExecutor extends BaseExecutor to provide the primary production backend for NeMo Curator pipelines. It leverages Cosmos-Xenna's mature streaming execution, autoscaling, and worker lifecycle management.

The executor does not support ignore_head_node and raises a ValueError if it is set to True, directing users to RayDataExecutor or RayActorPoolExecutor instead.

Initialization sets up default pipeline configuration:

• logging_interval: 60 seconds between status logs
• ignore_failures: False
• execution_mode: "streaming"
• cpu_allocation_percentage: 0.95
• autoscale_interval_s: 180 seconds

execute() follows this sequence:

1. Convert stages -- Each ProcessingStage is converted to a Xenna StageSpec by calling create_named_xenna_stage_adapter() and configuring worker counts, setup/run attempt limits, failure handling, worker lifetime, and other per-stage options from the stage's xenna_stage_spec().
2. Configure execution mode -- Selects streaming (default) or batch mode. For streaming mode, creates a StreamingSpecificSpec with autoscale interval and verbosity settings.
3. Build pipeline config -- Creates a PipelineConfig with execution mode, logging interval, failure handling, CPU allocation percentage, and verbosity levels.
4. Initialize Ray -- Calls ray.init() with RAY_EXPERIMENTAL_NOSET_CUDA_VISIBLE_DEVICES=0 so Xenna manages GPU assignment.
5. Run pipeline -- Calls pipelines_v1.run_pipeline() with the assembled PipelineSpec (input data, stages, config).
6. Cleanup -- Calls ray.shutdown() in a finally block.

User-provided configuration (via the config dict) is merged with defaults through _get_pipeline_config().

Usage

XennaExecutor is the default executor used by Pipeline.run() for production workloads. Use it when you need production-grade distributed execution with autoscaling, streaming mode, and worker lifecycle management. It requires the cosmos-xenna package.

Code Reference

Source Location

• Repository: NeMo-Curator
• File: nemo_curator/backends/xenna/executor.py
• Lines: 1-159

Signature

class XennaExecutor(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.xenna.executor import XennaExecutor

I/O Contract

Inputs

Name	Type	Required	Description
config	dict[str, Any] or None	No	Configuration dictionary (see Configuration Options below)
ignore_head_node	bool	No	Not supported; raises ValueError if True

execute() Inputs

Name	Type	Required	Description
stages	list[ProcessingStage]	Yes	List of processing stages to execute
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]	Output tasks from the last stage of the pipeline; empty list on failure with no results

Configuration Options

Key	Type	Default	Description
logging_interval	int	60	Seconds between status log messages
ignore_failures	bool	False	Whether to continue processing on individual task failures
execution_mode	str	"streaming"	Either "streaming" or "batch"
cpu_allocation_percentage	float	0.95	Fraction of available CPUs to allocate to workers
autoscale_interval_s	int	180	Seconds between autoscaling decisions (streaming mode only)

Per-Stage Configuration

Each stage can provide Xenna-specific configuration via its xenna_stage_spec() method. The following keys are supported in the returned dictionary:

Key	Description
num_workers	Fixed number of workers for this stage
num_workers_per_node	Number of workers per cluster node
num_setup_attempts_python	Max retry attempts for stage setup
num_run_attempts_python	Max retry attempts for processing
ignore_failures	Per-stage failure handling override
reset_workers_on_failure	Whether to reset worker actors after failures
slots_per_actor	Number of task slots per worker actor
worker_max_lifetime_m	Maximum worker lifetime in minutes
worker_restart_interval_m	Interval between worker restarts in minutes
max_setup_failure_percentage	Maximum allowed percentage of setup failures

Usage Examples

Running a Pipeline with XennaExecutor

from nemo_curator.backends.xenna.executor import XennaExecutor

# Create executor with custom configuration
executor = XennaExecutor(config={
    "logging_interval": 30,
    "execution_mode": "streaming",
    "cpu_allocation_percentage": 0.9,
    "autoscale_interval_s": 120,
})

# Execute the pipeline
results = executor.execute(
    stages=[reader_stage, processing_stage, writer_stage],
    initial_tasks=my_input_tasks,
)

Using Batch Execution Mode

from nemo_curator.backends.xenna.executor import XennaExecutor

executor = XennaExecutor(config={"execution_mode": "batch"})
results = executor.execute(stages=my_stages, initial_tasks=my_tasks)

Related Pages

• Environment:NVIDIA_NeMo_Curator_Python_Linux_Base
• NVIDIA_NeMo_Curator_Backend_Base_Classes -- Parent base classes
• NVIDIA_NeMo_Curator_XennaStageAdapter -- Adapter used to convert stages for Xenna
• NVIDIA_NeMo_Curator_RayDataExecutor -- Alternative experimental executor