API ReferenceFull Library ReferenceNemo CuratorNemo CuratorPipeline

nemo_curator.pipeline.pipeline

View as Markdown

Module Contents

Classes

NameDescription
PipelineUser-facing pipeline definition for composing processing stages.

Functions

NameDescription
assign_root_task_idsAssign root task_ids to user-provided initial tasks.

API

class nemo_curator.pipeline.pipeline.Pipeline(
    name: str,
    description: str | None = None,
    stages: list[nemo_curator.stages.base.ProcessingStage] | None = None,
    config: dict[str, typing.Any] | None = None
)

User-facing pipeline definition for composing processing stages.

config
= config or {}
stages
list[ProcessingStage] = stages or []
nemo_curator.pipeline.pipeline.Pipeline.__repr__() -> str

String representation of the pipeline.

nemo_curator.pipeline.pipeline.Pipeline._assign_source_sink_roles() -> None
nemo_curator.pipeline.pipeline.Pipeline._decompose_stages(
    stages: list[nemo_curator.stages.base.ProcessingStage | nemo_curator.stages.base.CompositeStage]
) -> tuple[list[nemo_curator.stages.base.ProcessingStage], dict[str, list[str]]]

Decompose composite stages into execution stages.

Parameters:

stages
list[ProcessingStage | CompositeStage]

List of stages that may include composite stages

Returns: tuple[list[ProcessingStage], dict[str, list[str]]]

tuple[list[ProcessingStage], dict[str, list[str]]]: Tuple of (execution stages, decomposition info dict)

Raises:

  • TypeError: If a composite stage is decomposed into another composite stage
nemo_curator.pipeline.pipeline.Pipeline._run_with_resumability(
    executor: nemo_curator.backends.base.BaseExecutor,
    initial_tasks: list[nemo_curator.tasks.Task] | None,
    checkpoint_path: pathlib.Path
) -> list[nemo_curator.tasks.Task] | None

Run with resumability around a pre-existing Ray cluster (e.g. one started by RayClient).

We briefly connect with with ray.init() to spawn the detached checkpoint actor, then disconnect before executor.execute so the executor’s own ray.init runs un-nested — a nested ray.init(runtime_env=...) is silently dropped, so the executor’s env vars wouldn’t propagate otherwise. The detached actor lives in the cluster across the executor’s separate Ray session; a final with ray.init() closes and kills it. The cluster must pre-exist: had we started it, the first with-exit shutdown would tear it down and take the actor with it.

Add a stage to the pipeline.

Parameters:

stage
ProcessingStage

Processing stage to add

Returns: Pipeline

Self (Pipeline) for method chaining

nemo_curator.pipeline.pipeline.Pipeline.build() -> None

Build an execution plan from the pipeline.

Raises:

  • ValueError: If the pipeline has no stages
nemo_curator.pipeline.pipeline.Pipeline.describe() -> str

Get a detailed description of the pipeline stages and their requirements.

nemo_curator.pipeline.pipeline.Pipeline.run(
    executor: nemo_curator.backends.base.BaseExecutor | None = None,
    initial_tasks: list[nemo_curator.tasks.Task] | None = None,
    checkpoint_path: str | pathlib.Path | None = None
) -> list[nemo_curator.tasks.Task] | None

Run the pipeline.

Parameters:

executor
BaseExecutorDefaults to None

Executor to use

initial_tasks
list[Task]Defaults to None

Initial tasks to start the pipeline with. Defaults to None.

checkpoint_path
str | PathDefaults to None

Resumability directory. Must be a LOCAL filesystem path (the LMDB state is written locally), not a remote/cloud URI. When set, completed source partitions are tracked (in a .nemo_curator_metadata subdir) and skipped on rerun. Multiple runs (e.g. a SLURM array) may share the directory — each writes its own LMDB file, so there is no contention.

Returns: list[Task] | None

list[Task] | None: List of tasks

nemo_curator.pipeline.pipeline.assign_root_task_ids(
    initial_tasks: list[nemo_curator.tasks.Task]
) -> list[nemo_curator.tasks.Task]

Assign root task_ids to user-provided initial tasks.

Every task in a run descends from the implicit root "0" (the id of :class:EmptyTask). User-provided initial tasks are its direct children, so they get "0_0", "0_1", … EmptyTask instances are skipped (already "0"). All downstream task_id assignment happens in BaseStageAdapter.

NOTE: we deliberately use the positional index here, NOT get_deterministic_id(), even for content-bearing tasks like FileGroupTask. The source stage is the single place content-based ids are assigned (to its outputs); hashing here too would put the content hash at two levels of the id path ("0_<hashA>_<hashB>"). Passing initial tasks directly is rare; if you need reorder-stable source ids, let a source stage emit them.