> This page is for version 26.07 · v1.3.0.
> For other versions, use one of these documentation indexes:
> - Latest · v1.4.0 (26.09) (default): https://docs.nvidia.com/nemo/curator/latest/llms.txt
> - Main · preview: https://docs.nvidia.com/nemo/curator/main/llms.txt
> - 26.09 · v1.4.0: https://docs.nvidia.com/nemo/curator/v26.09/llms.txt
> - 26.07 · v1.3.0: https://docs.nvidia.com/nemo/curator/v26.07/llms.txt
> - 26.04 · v1.2.0: https://docs.nvidia.com/nemo/curator/v26.04/llms.txt
> - 26.02 · v1.1.0: https://docs.nvidia.com/nemo/curator/v26.02/llms.txt
> - 25.09 · v1.0.0: https://docs.nvidia.com/nemo/curator/v25.09/llms.txt

> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.nvidia.com/nemo/curator/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.nvidia.com/nemo/curator/_mcp/server.

> Configure fixed and autoscaling worker pools and backend stage overrides for Ray Data, Xenna, and Ray Actor Pool executors

# Stage Worker Sizing

Use `ProcessingStage.num_workers()` or `stage.with_(num_workers=...)` to set a backend-neutral worker request. 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.
* `Resources` controls CPU and GPU resources reserved for each worker.
* `ray_stage_spec()` and `xenna_stage_spec()` provide backend-specific scheduling options.

> **Warning**
>
> A worker count does not create cluster capacity. Ensure the cluster can satisfy `worker count × per-worker Resources`, or use autoscaling bounds that fit the cluster.

## Common Stage for the Examples

The examples on this page configure this minimal stage:

```python
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

| Backend              | `num_workers=N`, where `N > 0`                                          | `num_workers=None`                                               | Backend-specific alternative                        |
| -------------------- | ----------------------------------------------------------------------- | ---------------------------------------------------------------- | --------------------------------------------------- |
| Ray Data actor stage | Fixed `ActorPoolStrategy(size=N)`                                       | Autoscaling actor pool, minimum 1 with no maximum by default     | `min_workers`, `max_workers`, and `initial_workers` |
| Ray Data task stage  | Fixed `TaskPoolStrategy(size=N)`                                        | Ray Data's default task scheduling; no explicit compute strategy | None; actor-pool sizing keys are ignored            |
| Xenna                | Exactly `N` workers across the cluster                                  | Xenna determines worker allocation                               | `num_workers_per_node` in `xenna_stage_spec`        |
| Ray Actor Pool       | Requests `N`; warns and caps it if fewer actors fit available resources | Chooses the smaller of work batches and resource capacity        | None                                                |

Ray Data and Ray Actor Pool also 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`; do not use zero or negative Xenna worker counts.

## 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:

```python
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`:

```python
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)`.

| Actor-pool key                     | String value        | Default when omitted | Meaning                                                               |
| ---------------------------------- | ------------------- | -------------------- | --------------------------------------------------------------------- |
| `RayStageSpecKeys.MIN_WORKERS`     | `"min_workers"`     | `1`                  | Minimum live actors                                                   |
| `RayStageSpecKeys.MAX_WORKERS`     | `"max_workers"`     | `None`               | Maximum live actors; `None` leaves the pool unbounded by this setting |
| `RayStageSpecKeys.INITIAL_WORKERS` | `"initial_workers"` | `None`               | Initial pool size selected by Ray Data                                |

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)`:

```python
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.

### Ray Data precedence and errors

| Configuration                                                                | Result                                                                             |
| ---------------------------------------------------------------------------- | ---------------------------------------------------------------------------------- |
| Actor stage with positive `num_workers` only                                 | Fixed actor pool                                                                   |
| Actor stage with positive `num_workers` plus min/max/initial keys            | Fixed actor pool wins; Curator logs a warning listing ignored sizing keys          |
| Actor stage without positive `num_workers`, no sizing keys                   | Autoscaling actor pool with `min_size=1`, `max_size=None`, and `initial_size=None` |
| Task stage with positive `num_workers`                                       | Fixed task pool                                                                    |
| Task stage with min/max/initial keys                                         | Keys are ignored; Curator logs a warning because they apply only to actor stages   |
| Invalid actor-pool bounds                                                    | Raises `ValueError`                                                                |
| Unknown `ray_stage_spec` key                                                 | Raises `ValueError` while constructing the adapter                                 |
| `ray_remote_args` includes `compute`, `max_calls`, `num_cpus`, or `num_gpus` | Raises `ValueError`; these arguments are managed by Curator                        |

## Xenna Worker Counts

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

#### Cluster-wide

Use the common worker hook. This example requests 12 workers in total, regardless of node count:

```python
from nemo_curator.backends.xenna import XennaExecutor

stage = PassthroughStage().with_(num_workers=12)
executor = XennaExecutor()
```

#### Per node

Leave `num_workers()` unset and pass `num_workers_per_node` through the Xenna stage spec:

```python
from nemo_curator.backends.xenna import XennaExecutor

stage = PassthroughStage().with_(
    num_workers=None,
    xenna_stage_spec={
        "num_workers_per_node": 2,
    },
)
executor = XennaExecutor()
```

The following combinations are invalid:

| Configuration                                                    | Behavior                                                              |
| ---------------------------------------------------------------- | --------------------------------------------------------------------- |
| `with_(xenna_stage_spec={"num_workers": 4})`                     | Raises `ValueError`; use `with_(num_workers=4)`                       |
| A stage's `xenna_stage_spec()` returns `{"num_workers": 4}`      | `XennaExecutor` raises `ValueError`; override `num_workers()` instead |
| Any non-`None` `num_workers()` result and `num_workers_per_node` | `XennaExecutor` raises `ValueError`; choose one scope                 |

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. If it 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:

```python
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`, and `num_workers` 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:

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

then this override:

```python
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`

| Call                                                    | Meaning                                                                 |
| ------------------------------------------------------- | ----------------------------------------------------------------------- |
| Omit `num_workers`                                      | Preserve the stage's current `num_workers()` result                     |
| `with_(num_workers=None)`                               | Explicitly reset worker sizing to executor-controlled behavior          |
| Omit `ray_stage_spec` or pass `ray_stage_spec=None`     | Do not modify the current Ray spec                                      |
| `ray_stage_spec={RayStageSpecKeys.MAX_WORKERS: None}`   | Preserve the key with an explicit `None`, clearing an inherited maximum |
| Omit `xenna_stage_spec` or pass `xenna_stage_spec=None` | Do not modify the current Xenna spec                                    |
| `xenna_stage_spec={"worker_max_lifetime_m": None}`      | Preserve that key with an explicit `None`                               |
| `runtime_env=None`                                      | Do not modify the current environment                                   |
| `runtime_env={}`                                        | Replace the current environment with an empty dictionary                |

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

## Reserved `num_workers` Hook

`num_workers` is reserved as a method on `ProcessingStage`. Do not define it as a dataclass field or ordinary stage attribute:

```python
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:

```python
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:

| Resource | Selection                                                                                                       |
| -------- | --------------------------------------------------------------------------------------------------------------- |
| CPU      | `ray_stage_spec[RayStageSpecKeys.RAY_NUM_CPUS]` when non-`None`; otherwise `stage.resources.cpus` when positive |
| GPU      | `stage.resources.gpus` when positive                                                                            |

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

```python
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.

> **Note**
>
> Lowering the scheduler reservation does not reduce the CPU work performed by FFmpeg. Confirm that the resulting concurrency does not oversubscribe the host.

## Related References

* [`ProcessingStage` API](/api/reference/api-reference/processing-stage)
* [Pipeline Execution Backends](/reference/infra/execution-backends)
* [Per-Stage Runtime Environments](/reference/infra/per-stage-runtime)
* [`Resources` API](/api/reference/api-reference/resources)