Stage Worker Sizing

Use ProcessingStage.num_workers() or num_workers_per_node() to set a backend-neutral worker request. The corresponding stage.with_(...) arguments configure one stage instance. Use ray_stage_spec and xenna_stage_spec for backend-specific execution behavior.

Worker count and per-worker resources are separate controls:

• num_workers() controls how many workers the executor requests.
• num_workers_per_node() scales a worker request by the alive Ray node count.
• Resources controls CPU and GPU resources reserved for each worker.
• ray_stage_spec() and xenna_stage_spec() provide backend-specific scheduling options.

Common Stage for the Examples

The examples on this page configure this minimal stage:

from nemo_curator.stages.base import ProcessingStage
from nemo_curator.stages.resources import Resources
from nemo_curator.tasks import EmptyTask
class PassthroughStage(ProcessingStage[EmptyTask, EmptyTask]):
    name = "passthrough"
    resources = Resources(cpus=1.0)
    def process(self, task: EmptyTask) -> EmptyTask:
        return task

Worker Controls by Backend

For the cluster-wide num_workers hook, Ray Data and Ray Actor Pool treat 0 and negative values as no fixed-size request. Xenna passes every non-None value to its stage specification, so use a positive count or None.

The common num_workers_per_node hook accepts positive integers or fractions. Curator rejects booleans, non-numeric values, zero, and negative values. It cannot be combined with num_workers or Ray Data’s min_workers, max_workers, and initial_workers keys.

Ray Data Worker Pools

Ray Data uses actor stages for stateful setup or GPU-backed work and task stages for stateless work. A stage is an actor stage when either condition is true:

• ray_stage_spec[RayStageSpecKeys.IS_ACTOR_STAGE] is explicitly True.
• The stage overrides setup(), or requests both CPU and GPU resources.

Set IS_ACTOR_STAGE to False to force task execution. Do this only when the stage does not depend on worker-local setup state.

Fixed actor pool

Set a positive num_workers and mark the stage as an actor stage:

from nemo_curator.backends.ray_data import RayDataExecutor
from nemo_curator.backends.utils import RayStageSpecKeys
stage = PassthroughStage().with_(
    num_workers=4,
    ray_stage_spec={
        RayStageSpecKeys.IS_ACTOR_STAGE: True,
    },
)
executor = RayDataExecutor()

Ray Data receives ActorPoolStrategy(size=4). Cluster resources must be able to place four actors with the stage’s per-worker Resources.

Autoscaling actor pool

Use the exact RayStageSpecKeys members MIN_WORKERS, MAX_WORKERS, and INITIAL_WORKERS:

from nemo_curator.backends.ray_data import RayDataExecutor
from nemo_curator.backends.utils import RayStageSpecKeys
stage = PassthroughStage().with_(
    # Explicitly clear an inherited fixed worker count, if the stage has one.
    num_workers=None,
    ray_stage_spec={
        RayStageSpecKeys.IS_ACTOR_STAGE: True,
        RayStageSpecKeys.MIN_WORKERS: 2,
        RayStageSpecKeys.MAX_WORKERS: 8,
        RayStageSpecKeys.INITIAL_WORKERS: 4,
    },
)
executor = RayDataExecutor()

Ray Data receives ActorPoolStrategy(min_size=2, max_size=8, initial_size=4).

Ray validates the relationship among the three values. For example, an initial size greater than the maximum raises ValueError with the stage name and the underlying actor-pool validation message.

Fixed task pool

For a stateless task stage, a positive worker count selects TaskPoolStrategy(size=N):

from nemo_curator.backends.ray_data import RayDataExecutor
from nemo_curator.backends.utils import RayStageSpecKeys
stage = PassthroughStage().with_(
    num_workers=6,
    ray_stage_spec={
        RayStageSpecKeys.IS_ACTOR_STAGE: False,
    },
)
executor = RayDataExecutor()

Without a positive num_workers, a task stage does not set compute; Ray Data uses its default task scheduler.

Pool sizing per node

Use the same portable override for actor or task stages:

stage = PassthroughStage().with_(num_workers_per_node=2)
executor = RayDataExecutor()

Ray Data pool sizes are cluster-wide, so Curator counts alive nodes and sets the pool size to ceil(num_workers_per_node × node count). For this per-node configuration, Curator sets scheduling_strategy="SPREAD" unless RAY_REMOTE_ARGS supplies another strategy.

SPREAD is placement intent, not a hard per-node cap. Ray may place workers unevenly when resources differ or placement constraints apply.

Ray Data precedence and errors

Xenna Worker Counts

Xenna supports either a cluster-wide fixed count or a per-node count, but not both on one stage.

from nemo_curator.backends.xenna import XennaExecutor
stage = PassthroughStage().with_(num_workers=12)
executor = XennaExecutor()

The following combinations are invalid:

The legacy xenna_stage_spec={"num_workers_per_node": N} form remains supported for compatibility. When neither setting is present, Xenna controls worker allocation using stage resources and pipeline demand.

Ray Actor Pool Resource Capping

RayActorPoolExecutor calculates how many actors fit from available CPU and GPU resources. A positive num_workers() is honored when it fits. num_workers_per_node() first multiplies the value by the alive node count and uses best-effort SPREAD placement. If either request exceeds capacity, Curator warns and caps the actor count instead of waiting for impossible placements.

For example, a stage that requests Resources(cpus=2.0) and num_workers=8 can place at most four actors when eight CPUs are available. Work batch count can further reduce the automatically chosen pool size when num_workers is unset.

Configure a Stage with with_()

with_() deep-copies the stage and returns the configured copy. It can override portable properties and merge backend stage specs in one call:

from nemo_curator.backends.utils import RayStageSpecKeys
from nemo_curator.stages.resources import Resources
configured = PassthroughStage().with_(
    name="gpu_passthrough",
    resources=Resources(cpus=2.0, gpus=1.0),
    batch_size=8,
    runtime_env={"pip": ["transformers==4.45.0"]},
    num_workers=4,
    ray_stage_spec={
        RayStageSpecKeys.RAY_NUM_CPUS: 1.0,
        RayStageSpecKeys.RAY_REMOTE_ARGS: {
            "scheduling_strategy": "SPREAD",
        },
    },
    xenna_stage_spec={
        "worker_max_lifetime_m": 45,
    },
)

The original stage remains unchanged. resources, batch_size, runtime_env, num_workers, and num_workers_per_node apply across supported executors. ray_stage_spec configures Ray Data and the Ray-specific flags used by Ray Actor Pool; xenna_stage_spec configures Xenna.

Shallow-merge behavior

ray_stage_spec and xenna_stage_spec are shallow-merged with the stage’s existing dictionary. New top-level keys are preserved, and user values replace existing values at the same top-level key. Nested dictionaries are replaced, not recursively merged.

For example, if a stage returns:

{
    RayStageSpecKeys.MAX_WORKERS: 8,
    RayStageSpecKeys.RAY_REMOTE_ARGS: {
        "scheduling_strategy": "SPREAD",
        "max_retries": 2,
    },
}

then this override:

stage.with_(
    ray_stage_spec={
        RayStageSpecKeys.MAX_WORKERS: None,
        RayStageSpecKeys.RAY_REMOTE_ARGS: {"max_retries": 0},
    }
)

produces a spec where max_workers is explicitly None and ray_remote_args contains only {"max_retries": 0}. The original scheduling_strategy nested key is not retained.

Omitted values and explicit None

Chained with_() calls merge against the already configured copy, so later values win while unrelated top-level keys from earlier calls remain.

Reserved Worker-Sizing Hooks

num_workers and num_workers_per_node are reserved as methods on ProcessingStage. Do not define either as a dataclass field or ordinary stage attribute:

class FixedStage(PassthroughStage):
    def num_workers(self) -> int:
        return 4

Alternatively, keep the class executor-neutral and use stage.with_(num_workers=4) at pipeline construction time. Defining num_workers: int = 4 as a class or dataclass field raises TypeError when the subclass is created.

Some built-in single-input and fanout stages override num_workers() to return 1, preventing duplicate source enumeration. Use with_(num_workers=None) only when executor-controlled parallelism is safe for that stage’s input contract.

For ASR stages, the stage-local NeMo DataLoader setting is named dataloader_num_workers. When migrating an ASR constructor or configuration, rename the previous stage field num_workers to dataloader_num_workers; the common num_workers() hook is now reserved for Curator backend workers:

from nemo_curator.stages.audio.tagging.inference.nemo_asr_align import NeMoASRAlignerStage
asr_stage = NeMoASRAlignerStage(
    dataloader_num_workers=10,
).with_(num_workers=2)

Resource Matching and Ray Data Fusion

Ray Data may fuse compatible adjacent operators to avoid materializing and serializing intermediate blocks. Fusion is an optimizer decision, not a guaranteed stage behavior.

Ray Data determines the per-worker reservation in this order:

Use RAY_NUM_CPUS when Ray Data needs a different scheduling reservation from other executors:

stage = PassthroughStage().with_(
    resources=Resources(cpus=4.0),
    ray_stage_spec={
        RayStageSpecKeys.RAY_NUM_CPUS: 1.0,
    },
)

The CPU paths of VideoFrameExtractionStage and ClipTranscodingStage use a Ray Data reservation of 1.0 CPU by default while preserving their larger Resources.cpus values for Xenna. Matching the adjacent video operators’ Ray Data reservations makes fusion possible and can remove inter-stage serialization overhead. Different resources, execution constraints, or optimizer decisions can still keep stages separate; inspect the Ray Data execution plan or dashboard instead of assuming fusion occurred.

Related References

• ProcessingStage API
• Pipeline Execution Backends
• Per-Stage Runtime Environments
• Resources API