XennaExecutor

1
from nemo_curator.backends.xenna import XennaExecutor

1
class XennaExecutor(BaseExecutor):
2
    """Production executor using Cosmos-Xenna for distributed execution.
3
4
    Provides:
5
    - Distributed task orchestration
6
    - Resource allocation and management
7
    - Batch processing optimization
8
    - Performance metrics collection
9
    """
10
11
    def __init__(
12
        self,
13
        config: dict[str, Any] | None = None,
14
        ignore_head_node: bool = False,
15
    ) -> None:
16
        """Initialize the executor.
17
18
        Args:
19
            config: Executor configuration dictionary.
20
            ignore_head_node: Not supported. Raises ValueError if True.
21
        """

1
from nemo_curator.pipeline import Pipeline
2
from nemo_curator.backends.xenna import XennaExecutor
3
4
pipeline = Pipeline(name="my_pipeline", stages=[...])
5
6
# Default executor
7
results = pipeline.run()
8
9
# Equivalent to:
10
executor = XennaExecutor()
11
results = pipeline.run(executor=executor)

1
from nemo_curator.stages.base import ProcessingStage
2
from nemo_curator.stages.resources import Resources
3
from nemo_curator.tasks import EmptyTask
4
5
6
class PassthroughStage(ProcessingStage[EmptyTask, EmptyTask]):
7
    name = "passthrough"
8
    resources = Resources(cpus=1.0)
9
10
    def process(self, task: EmptyTask) -> EmptyTask:
11
        return task
12
13
14
cluster_wide_stage = PassthroughStage().with_(
15
    num_workers=12,
16
    xenna_stage_spec={
17
        "worker_max_lifetime_m": 45,
18
    },
19
)
20
21
per_node_stage = PassthroughStage().with_(
22
    num_workers=None,
23
    xenna_stage_spec={
24
        "num_workers_per_node": 2,
25
    },
26
)

1
executor = XennaExecutor(config={
2
    "logging_interval": 30,
3
    "ignore_failures": True,
4
    "execution_mode": "batch",
5
    "cpu_allocation_percentage": 0.9,
6
})
7
8
results = pipeline.run(executor=executor)

1
def execute(
2
    self,
3
    stages: list[ProcessingStage],
4
    initial_tasks: list[Task] | None = None,
5
) -> list[Task]:
6
    """Execute the pipeline stages.
7
8
    Args:
9
        stages: List of processing stages to execute.
10
        initial_tasks: Initial tasks (defaults to a fresh `EmptyTask()` instance).
11
12
    Returns:
13
        List of output tasks from the final stage.
14
    """

1
executor = XennaExecutor(config={
2
    "ignore_failures": True,  # Continue despite errors
3
})
4
5
try:
6
    results = pipeline.run(executor=executor)
7
except Exception as e:
8
    # Handle pipeline-level failures
9
    print(f"Pipeline failed: {e}")

1
results = pipeline.run(executor=executor)
2
3
# Each task contains performance data
4
for task in results:
5
    for perf in task._stage_perf:
6
        print(f"Stage: {perf.stage_name}")
7
        print(f"  Duration: {perf.process_time}s")
8
        print(f"  Items processed: {perf.num_items_processed}")