wehub-resource-sync
107 lines
5.1 KiB
Markdown
107 lines
5.1 KiB
Markdown
(data_concurrent_execution)=
|
|
|
|
# Run multiple Datasets in one cluster
|
|
|
|
When two or more Ray Data Datasets share a single Ray cluster, they compete for the same pool of nodes by default. That competition can cause unwanted contention — one Dataset's reads can starve a second Dataset's GPU stage, autoscaling decisions get muddled, and runtime becomes a function of whatever else happens to be running.
|
|
|
|
Ray Data lets you assign each Dataset to its own **subcluster** — a labeled subset of nodes that only that Dataset uses. Subclusters give you smooth, predictable execution for concurrent Datasets and a natural way to express "this Dataset runs here, that one runs there."
|
|
|
|
Common use cases:
|
|
|
|
- **Asynchronous validation during training.** A training Dataset feeds the trainer. A validation Dataset feeds a separate validation task on different hardware. See {ref}`train-validating-checkpoints` for the Ray Train integration.
|
|
- **Multitenancy on a shared workspace.** Several Datasets — different users, different pipelines, or different stages of one workflow — share one Anyscale workspace and don't disturb each other.
|
|
|
|
## How it works
|
|
|
|
Each Dataset carries an `ExecutionOptions.label_selector` (a `Dict[str, str]`) that Ray Data attaches to every task and actor the Dataset launches. The autoscaling coordinator buckets nodes by the value at the reserved label key `"ray-subcluster"` and only places a Dataset's work on nodes whose label matches.
|
|
|
|
## Configuration
|
|
|
|
There are two steps.
|
|
|
|
### 1. Label your worker nodes
|
|
|
|
Label each worker node with the reserved key `ray-subcluster` to mark which subcluster it belongs to. See {ref}`labels` for how to configure labels. The mechanism used depends on your deployment (cluster YAML, KubeRay, or `ray start --labels`).
|
|
|
|
For example, in a Ray cluster YAML config:
|
|
|
|
```yaml
|
|
available_node_types:
|
|
train_workers:
|
|
min_workers: 2
|
|
max_workers: 4
|
|
labels:
|
|
ray-subcluster: training
|
|
node_config:
|
|
InstanceType: g5.xlarge
|
|
validation_workers:
|
|
min_workers: 0
|
|
max_workers: 2
|
|
labels:
|
|
ray-subcluster: validation
|
|
node_config:
|
|
InstanceType: g4dn.xlarge
|
|
```
|
|
|
|
Subcluster values are arbitrary strings (`"training"`, `"validation"`, `"tenant_a"`, `"team-blue"`) — pick whatever makes sense for your workload.
|
|
|
|
### 2. Tag each Dataset with a `label_selector`
|
|
|
|
Copy the current `DataContext`, set the selector on the copy, and apply the copy temporarily with the `DataContext.current()` context manager. Construct your Dataset inside the `with` block:
|
|
|
|
```python
|
|
import ray
|
|
|
|
ctx = ray.data.DataContext.get_current().copy()
|
|
ctx.execution_options.label_selector = {"ray-subcluster": "tenant_a"}
|
|
|
|
with ray.data.DataContext.current(ctx):
|
|
# Tasks launched during construction (reads, schema inference) read
|
|
# the temporary context. ``Dataset.context`` is a deep copy of the
|
|
# current context, so the new Dataset keeps the selector after the
|
|
# ``with`` block exits.
|
|
dataset = ray.data.read_parquet("s3://my-bucket/tenant_a/")
|
|
```
|
|
|
|
:::{important}
|
|
Mutating `ray.data.DataContext.get_current()` in place permanently affects every subsequent Dataset in the same driver process. Use the `DataContext.current()` context manager to scope each Dataset's selector to its own construction block.
|
|
|
|
Set the selector *before* creating the Dataset, not after. Tasks Ray Data spawns during construction (for example, the parquet read tasks that infer the schema) read the current context, so setting `dataset.context.execution_options.label_selector` after the fact doesn't retroactively re-route them.
|
|
:::
|
|
|
|
## Example: two Datasets, two subclusters
|
|
|
|
```python
|
|
import ray
|
|
import threading
|
|
|
|
|
|
def make_dataset(subcluster: str, path: str) -> ray.data.Dataset:
|
|
ctx = ray.data.DataContext.get_current().copy()
|
|
ctx.execution_options.label_selector = {"ray-subcluster": subcluster}
|
|
with ray.data.DataContext.current(ctx):
|
|
return ray.data.read_parquet(path)
|
|
|
|
|
|
# Construct each Dataset in the main thread so the temporary contexts
|
|
# don't race on the process-global ``_default_context``.
|
|
ds_a = make_dataset("tenant_a", "s3://my-bucket/tenant_a/")
|
|
ds_b = make_dataset("tenant_b", "s3://my-bucket/tenant_b/")
|
|
|
|
# Then run them concurrently. ds_a's tasks only land on
|
|
# ray-subcluster=tenant_a nodes; ds_b's only on
|
|
# ray-subcluster=tenant_b nodes.
|
|
threading.Thread(target=lambda: ds_a.materialize()).start()
|
|
threading.Thread(target=lambda: ds_b.materialize()).start()
|
|
```
|
|
|
|
## Ray Train integration
|
|
|
|
When you wire the Datasets into a `TorchTrainer` (or any `DataParallelTrainer`), `ray.train.DataConfig` is the more ergonomic entry point — it takes a per-dataset `ExecutionOptions` map. See {ref}`train-validating-checkpoints` for the full pattern, including how to set the training-side selector through `DataConfig` and the validation-side selector inside your `validation_fn`.
|
|
|
|
## API reference
|
|
|
|
- {class}`ray.data.ExecutionOptions` — see the `label_selector` parameter.
|
|
- {class}`ray.data.DataContext` — the per-process Ray Data configuration the `execution_options` live on.
|
|
- {class}`ray.train.DataConfig` — accepts a `Dict[str, ExecutionOptions]` so each Train dataset can carry its own selector.
|