Repository navigation
docs: address 9 open documentation issues #1722
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
8b6d3c0
146d52e
a60bfe0
ad119d2
08bfaba
f063ae3
3442ec4
c48097c
a415a65
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Can we get the image without the red underline on DocumentBatch? |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| --- | ||
| description: "How NeMo Curator automatically balances resources across pipeline stages" | ||
| categories: ["architecture"] | ||
| tags: ["deep-dive", "auto-balancing", "scheduling", "performance"] | ||
| personas: ["data-scientist-focused", "mle-focused"] | ||
| difficulty: "intermediate" | ||
| content_type: "concept" | ||
| modality: "universal" | ||
| --- | ||
|
|
||
| # Auto-Balancing Heterogeneous Models | ||
|
|
||
| NeMo Curator auto-balances resources at the application level across pipeline stages to maximize throughput. This means you can focus on defining your curation logic rather than manually tuning parallelism. | ||
|
|
||
| ## The Problem: Unbalanced Pipelines | ||
|
|
||
| In a typical curation pipeline, stages have very different processing speeds. Consider a video pipeline with a fast stage, a slow stage, and a medium stage — all sharing a fixed GPU budget: | ||
|
|
||
| **Without auto-balancing (3 GPUs, 1 worker per stage):** | ||
| - The **fast stage** emits 4 tasks/s, but the **slow stage** can only handle 1 task/s. Jobs back up in the queue — 3 tasks accumulate per second, eventually causing memory pressure. | ||
| - The **medium stage** produces 2 tasks/s but is limited by the slow stage upstream, so it's starved for work. In practice only ~1 task/s is realized. | ||
| - 4 queued tasks per second could have been processed, but the final stage is starved. | ||
|
|
||
| **With auto-balancing (7 GPUs, scaled workers):** | ||
| - The executor detects the bottleneck and scales the **slow stage to 4× workers** and the **medium stage to 2× workers**. | ||
| - Now every stage sustains **4 tasks/s** throughput. Queues stay relatively clear — new jobs are picked up promptly. | ||
| - **Result: 4× throughput improvement** by intelligently redistributing the same GPU budget. | ||
|
|
||
| ## How Auto-Balancing Works | ||
|
|
||
| The executor monitors the throughput and queue depth of each stage at runtime and uses this information to: | ||
|
|
||
| 1. **Monitor throughput of different stages** and rebalance resources at regular intervals, shifting GPU/CPU allocations toward bottleneck stages. | ||
| 2. **Apply backpressure.** When a downstream stage can't keep up, upstream stages slow their output rate rather than buffering unbounded data in memory. This reduces memory pressure and prevents spilling to disk. | ||
| 3. **Scale workers dynamically.** If a stage is falling behind, the executor allocates additional workers to that stage (within the available resource budget). | ||
|
|
||
| ## What This Means for You | ||
|
|
||
| - **No manual parallelism tuning.** You don't need to calculate the optimal number of workers per stage — the executor adapts at runtime. | ||
| - **Predictable memory usage.** Backpressure prevents unbounded buffering, so memory usage stays stable even with unbalanced stages. | ||
| - **Efficient hardware utilization.** Resources shift toward the current bottleneck instead of being statically allocated. | ||
|
|
||
| ## Monitoring Stage Balance | ||
|
|
||
| Use the Ray Dashboard to monitor how the executor is balancing your pipeline. If you notice a persistent bottleneck that auto-balancing can't resolve (for example, a stage that needs more GPU memory than is available), consider splitting the pipeline or scaling your cluster. |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| --- | ||
| description: "How NeMo Curator allocates CPUs, GPUs, and memory across pipeline stages" | ||
| categories: ["architecture"] | ||
| tags: ["deep-dive", "resources", "gpu", "cpu", "allocation"] | ||
| personas: ["data-scientist-focused", "mle-focused"] | ||
| difficulty: "intermediate" | ||
| content_type: "concept" | ||
| modality: "universal" | ||
| --- | ||
|
|
||
| # Scaling Up with Ray Resource Allocation | ||
|
|
||
| NeMo Curator makes resource allocation across pipeline stages straightforward — both CPUs and GPUs. Each stage declares its own resource requirements, and the executor schedules work accordingly. | ||
|
|
||
| This design improves performance on CPU-only stages in pipelines that also use GPUs, because CPU stages no longer block GPU resources. | ||
|
|
||
| ## How It Works | ||
|
|
||
| Every `ProcessingStage` can specify a `Resources` object that declares its CPU and GPU needs: | ||
|
|
||
| ```python | ||
| from nemo_curator.stages.core import ProcessingStage | ||
| from nemo_curator.stages.function_definitions import processing_stage | ||
| from nemo_curator.stages.resources import Resources | ||
| from nemo_curator.tasks import DocumentBatch | ||
|
|
||
|
|
||
| class TokenizerStage(ProcessingStage[DocumentBatch, DocumentBatch]): | ||
| name: str = "TokenizerStage" | ||
| resources: Resources = Resources(cpus=1.0) # CPU-only — no GPU needed | ||
|
|
||
| def __init__(self): | ||
| super().__init__() | ||
| # ... stage logic ... | ||
|
Comment on lines
+28
to
+34
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Nit but we don't want users setting the |
||
|
|
||
|
|
||
| class ModelStage(ProcessingStage[DocumentBatch, DocumentBatch]): | ||
| name: str = "ModelStage" | ||
|
|
||
| def __init__(self, model_path: str): | ||
| super().__init__() | ||
| # ... stage logic ... | ||
| pass | ||
|
|
||
|
|
||
| @processing_stage(name="custom_filter", resources=Resources(cpus=1)) | ||
| def custom_filter_stage(task: DocumentBatch) -> DocumentBatch: | ||
| # ... filter logic ... | ||
| pass | ||
|
|
||
|
|
||
| model_stage = ModelStage(model_path="path/to/model").with_(resources=Resources(gpus=1)) | ||
| ``` | ||
|
|
||
| When a pipeline runs, the executor reads each stage's resource declaration and schedules tasks to satisfy those constraints. Stages that need GPUs are placed on GPU-equipped nodes; CPU-only stages can run on any available worker. | ||
|
|
||
| ## Key Concepts | ||
|
|
||
| ### CPU-Only vs. GPU Stages | ||
|
|
||
| The most impactful optimization is correctly separating CPU and GPU work. In a mixed pipeline, CPU-only stages (tokenization, text parsing, filtering) should not request GPU resources — this frees GPUs for inference stages that actually need them: | ||
|
|
||
| ```python | ||
| # CPU-only: tokenization, filtering, I/O | ||
| # Runs on any worker, doesn't block GPU resources | ||
| @processing_stage(name="tokenizer", resources=Resources(cpus=1)) | ||
| def tokenizer_stage(task: DocumentBatch) -> DocumentBatch: | ||
| pass | ||
|
|
||
| # GPU: model inference, embeddings | ||
| # Scheduled only on GPU-equipped nodes | ||
| model_stage = ModelStage(model_path="path/to/model").with_(resources=Resources(gpus=1)) | ||
| ``` | ||
|
|
||
| ### Fractional GPU Allocation | ||
|
|
||
| Some GPU stages don't need an entire GPU. You can use fractional allocation via `Resources(gpus=0.25)` or reserve a specific amount of GPU memory with `Resources(gpu_memory_gb=10)`: | ||
|
|
||
| ```python | ||
| # 4 workers share one GPU via fractional allocation | ||
| model_stage = ModelStage(model_path="path/to/model").with_(resources=Resources(gpus=0.25)) | ||
|
|
||
| # Or reserve a specific amount of GPU memory | ||
| model_stage = ModelStage(model_path="path/to/model").with_(resources=Resources(gpu_memory_gb=10)) | ||
| ``` | ||
|
|
||
| This is useful for inference stages where the model fits in a fraction of GPU memory, allowing you to increase parallelism without requiring more hardware. Note that `gpu_memory_gb` sets GPU memory for a single GPU. | ||
|
|
||
| ## Best Practices | ||
|
|
||
| - **Start with defaults.** Most stages have sensible default resource declarations. Override only when you observe resource contention or underutilization. | ||
| - **Separate CPU and GPU stages.** This is the single highest-impact optimization — it allows the executor to parallelize across heterogeneous hardware. | ||
| - **Profile before tuning.** Use Ray Dashboard or stage performance stats to identify bottlenecks before adjusting allocations. | ||
| - **Match hardware to workload.** If your pipeline is mostly CPU-bound (text filtering), you may not need GPU nodes at all. | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| --- | ||
| description: "How NeMo Curator's streaming architecture processes data in constant memory" | ||
| categories: ["architecture"] | ||
| tags: ["deep-dive", "streaming", "batching", "memory", "data-flow"] | ||
| personas: ["data-scientist-focused", "mle-focused"] | ||
| difficulty: "intermediate" | ||
| content_type: "concept" | ||
| modality: "universal" | ||
| --- | ||
|
|
||
| # Streaming vs. Batch Processing | ||
|
|
||
| Different inference stages have different compute requirements. NeMo Curator uses Ray streaming to increase GPU utilization and processing speed compared to traditional batch-all-at-once approaches. | ||
|
|
||
| ## Batch Mode vs. Streaming Mode | ||
|
|
||
| ### Batch Mode | ||
|
|
||
| In batch mode, each stage processes the **entire dataset** before the next stage begins. Stages with different compute requirements (CPU-only tokenization, single-GPU classifiers, multi-GPU encoders) all run sequentially: | ||
|
|
||
| | Aspect | Batch Mode | | ||
| |--------|-----------| | ||
| | **Execution** | One stage at a time across the full dataset | | ||
| | **Memory usage** | Proportional to dataset size | | ||
| | **GPU utilization** | Low — GPUs idle while CPU stages run, and vice versa | | ||
| | **Time to first output** | After the entire pipeline finishes | | ||
|
|
||
| ### Streaming Mode | ||
|
|
||
| In streaming mode, data flows through the pipeline as discrete batches. Each stage processes its current batch and immediately passes it downstream, so **all stages run concurrently** on different batches: | ||
|
|
||
| | Aspect | Streaming Mode | | ||
| |--------|---------------| | ||
| | **Execution** | All stages active simultaneously on different batches | | ||
| | **Memory usage** | Constant (proportional to batch size, not dataset size) | | ||
| | **GPU utilization** | High — stages with different hardware needs overlap | | ||
| | **Time to first output** | After the first batch completes the pipeline | | ||
|
|
||
| ## Why Streaming Is Faster | ||
|
|
||
| Streaming with heterogeneous compute allows NeMo Curator to overlap stages that use different resources. For example, while a GPU inference stage processes batch N, a CPU tokenization stage can process batch N+1 simultaneously — neither blocks the other. | ||
|
|
||
| This overlap improves throughput in pipelines that mix CPU and GPU work, because both happen in parallel rather than taking turns. | ||
|
|
||
| Combined with auto-balancing, streaming enables Curator to rearrange resources so that GPU stage workers are **kept busy over 99% of the time** after an initial warm-up period. | ||
|
|
||
| ## Heterogeneous Executors | ||
|
|
||
| NeMo Curator supports streaming with multiple executors — Cosmos Xenna, Ray Data, and others — each optimized for different workload patterns. The executor handles scheduling, backpressure, and resource allocation so that streaming "just works" regardless of how many stages your pipeline has. | ||
|
|
||
| ## Configuring Batch Size | ||
|
|
||
| Batch size controls the trade-off between memory usage and throughput: | ||
|
|
||
| ```python | ||
| from nemo_curator.stages.resources import Resources | ||
|
|
||
| # Configure batch size on a stage | ||
| word_count_stage = WordCountStage().with_(batch_size=128) | ||
| ``` | ||
|
|
||
| - **Smaller batches**: Lower memory usage per batch. Ray may handle smaller batches more efficiently in some workloads. | ||
| - **Larger batches**: More in-memory data per batch, which can reduce I/O overhead but uses more memory. | ||
|
|
||
| ## When Streaming Matters Most | ||
|
|
||
| - **Datasets exceed memory.** A 10 TB Common Crawl snapshot cannot fit in RAM, but it can be streamed in manageable chunks. | ||
| - **Pipeline stages have different hardware needs.** CPU-only and GPU stages overlap instead of taking turns. | ||
| - **You need early results.** Inspect output from the first batch while the rest of the dataset is still processing. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This link goes to
NVIDIA/NeMo(the NeMo framework repo), notNVIDIA-NeMo/Curator. Contributors following it will land on the NeMo framework's contribution guide, which has different requirements. The correct URL is used in the PR checklist itself.