Ray integration

Ray is an optional parallelism / execution backend, not the Framework’s identity. Domain processors and algorithm APIs are identical for inprocess, local processes, and Ray. Alternative backends register through the parallel transport registry; see Compare for how this layer relates to Spark and orchestration.

Configuration

params_defaults:
  runtime.parallelism.backend: ray
  runtime.parallelism.workers: 4
  runtime.parallelism.ray_address: ray://ray-head:10001
  # optional:
  # runtime.parallelism.schedule: pull
  # runtime.parallelism.deflate_idle: true
  # runtime.parallelism.num_gpus: 0
  # runtime.parallelism.require_object_store: false

If the address is empty, RAY_ADDRESS is used; otherwise Ray starts locally. integration/ray/connect.py lazily imports Ray and calls ray.init once. When kubectl can reach the cluster, resolve_ray_client_address prefers the live head Service LoadBalancer over a stale RAY_ADDRESS (common after RayCluster / LB recreate).

Param / env	Role
runtime.parallelism.workers	Partition shard workers or embed_train W shards (Ray actors)
runtime.parallelism.ray_address / RAY_ADDRESS	Client address (ray://…); empty → local ray.init
runtime.parallelism.num_gpus	Per-actor GPU reservation (@ray.remote(num_gpus=…))
runtime.parallelism.cpus_per_worker / DRTOLLER_RAY_CPUS_PER_WORKER	Advertised vCPU per KubeRay worker pod (default 4)
runtime.parallelism.max_nodes / DRTOLLER_RAY_MAX_NODES	Cluster node cap including head (default 23 → maxReplicas = 22)
runtime.parallelism.require_object_store / DRTOLLER_REQUIRE_OBJECT_STORE	Force object-store guard even for single-worker / local heuristics
runtime.parallelism.schedule	static or pull (workers lease the next shard). Default static
runtime.parallelism.deflate_idle	With Ray + pull: shrink worker pods as actors finish. Default false

runtime.parallelism.* controls execution workers. compute.parallelism.* controls algorithm-internal resources and is separate.

Status and preflight

Before a partition Ray run, the backend calls preflight_ray (at least one alive node). Operators can inspect the cluster:

drtoller ray status
drtoller ray status --address ray://ray-head:10001 --format json

Prints alive/total nodes, CPU/GPU totals, and per-node resources.

Object-store guard (multi-host)

Workers must see the same parquet object keys. Host-local bindings (storage.stores.*.backend: local) break when actors run on other machines.

storage/ray_compat.py raises when local parquet bindings are present and any of:

• runtime.parallelism.require_object_store / DRTOLLER_REQUIRE_OBJECT_STORE=1; or
• ray_address is set and workers > 1 (compose / KubeRay-style topology).

Fix: use parquet backend: s3 (MinIO, AWS S3, or GCS S3-interop). Single-node local Ray tests may keep backend: local.

The same guard rejects a host-path corpus under multi-worker / remote Ray. Step01 must stream from object store (domain.corpus_from_s3=true + domain.step00_run_id → s3://raw/corpora/{run_id}/docs/). See Step01.

Cloud env presets (dotenv / JSON):

drtoller deploy env --list-providers
drtoller deploy env --provider minio
drtoller deploy env --provider aws_s3 --format json
drtoller deploy env --provider gcp_gcs

Platform env.example documents MinIO (current reference) and optional AWS/GCP key blocks for later. Airflow in that reference receives S3 credentials for parquet backend: s3. See Reference Compose environment.

Multi-host KubeRay (DigitalOcean)

Current supported path: VM compose (drtoller up) + existing DOKS.

# Profile bring-up (kubeconfig, pool limits, bootstrap, attach):
drtoller k8s up drtoller-4cpu8ram.env
drtoller verify

# Tear down Ray + shrink the pool (DOKS cluster stays):
drtoller k8s down drtoller-4cpu8ram.env

Split form: drtoller cloud bootstrap --provider digitalocean --attach (needs KUBECONFIG and .env MINIO_API_BIND / DIGITALOCEAN_ACCESS_TOKEN).

• ray attach writes RAY_ADDRESS=ray://<head-LB>:10001, stages host kubectl/doctl into ${DATADIR}/kube (bin/ on PATH, doctl/config.yaml), recreates Airflow (/opt/kube). k8s up waits until head/worker pods are Ready and the LB has an EXTERNAL-IP.
• Parquet stores for pipeline steps use backend: s3 (MinIO). Postgres/Qdrant unchanged.
• Creating the DOKS cluster itself is still a one-time cloud UI/doctl step; bootstrap installs workload (operator, secrets, RayCluster) and attaches Airflow.

Live cluster shape (UI → kubectl)

Streamlit parallelism_controls writes runtime.parallelism.cpus_per_worker and max_nodes. On a Ray task start, Framework ensure_raycluster_shape (cluster_shape.py packing mutate + kubectl in cluster_tune.py) patches the live RayCluster:

• maxReplicas = max_nodes − 1 (one slot for the head);
• worker pod CPU request/limit from cpus_per_worker (DOKS s-4vcpu-8gb: request 3, limit 4);
• object-store + /dev/shm capped at 128 MiB so plasma does not eat the spaCy cgroup;
• BLAS env (OMP_NUM_THREADS=1, …) so several actors can share a node.

Missing kubectl → log + skip; the run still starts. Host drtoller ray sync / Framework drtoller ray cluster-sync are debug-only. kubectl prefers /opt/kube/bin so doctl auth can live under /opt/kube/doctl.

In-tree autoscaling is on (enableInTreeAutoscaling). For map-style jobs, worker pods ≈ ceil(actors / cpus_per_worker), capped by maxReplicas. runtime.parallelism.workers is the actor count, not the node count. Inflate waits for packed worker CPUs (pods × cpus_per_worker), not the raw actor request — if maxReplicas already caps the cluster, actors are capped to advertised CPUs so the job can start.

partition_loop pins replicas/minReplicas to that packed pod count for the run, then JSON-patches idle (1/1) when the job finishes.

Sticky cluster fit is exclusive: one FitShardActor per pod (num_cpus=cpus_per_worker). Before loading vectors, Framework inflates replicas/minReplicas, waits for packed worker CPUs (runtime.parallelism.inflate_timeout_s / inflate_poll_s), then loads. Before nested evaluate, release() kills fit actors (cluster stays inflated). close() restores idle workers (replicas/minReplicas = 1) only at outermost probe depth. See Streaming cluster.

Not in scope yet: one-command AWS/EKS or GCP/GKE (backlog). Env presets aws_s3 / gcp_gcs exist for object store only.

partition_loop Ray backend

parallel/backends/ray_map.py:

1. patches live KubeRay packing, pins replicas/minReplicas to packed pods, waits for packed worker CPUs, then preflight;
2. asserts storage bindings are Ray-compatible (object-store guard);
3. creates a serializable WorkerRunSpec from manifest path, RunContext identity and params;
4. assigns a distinct write shard id per worker;
5. creates one actor per worker (optional num_gpus; actor count capped to advertised CPUs);
6. static: passes a fixed shard bag to each actor. pull: actors lease one shard at a time from a queue on the Ray head;
7. each actor constructs its own WorkerSession, storage session and processor context;
8. driver merges worker manifests and metric deltas; idle replica patch on exit. With deflate_idle, replicas drop as pull workers finish.

Workers stream primary shards (bag or leased). Corpus GET/read skips keys not in the worker shard_scope. Postgres sinks enqueue to a single writer actor. The driver does not ship full datasets or live StorageSession objects through Ray.

Worker and driver logs include phase timings: pipeline vs domain on the worker, PG writer vs storage close on finalize, MinIO finalize_part upload_ms, driver wait p50/p95. Use them to see whether a slow run is compute, Postgres, or object-store upload.

Metrics channel

When metrics are enabled, workers send neutral metric deltas through a Ray channel. Driver-side telemetry owns Prometheus emission. Worker code does not start scrape servers.

reduce_merge on Ray/processes uses the same channel (processing/reduce_merge/live_metrics.py): workers push WorkerMetricDelta; a driver heartbeat (~5 s) ticks Grafana so panels are not mute while map_jobs blocks. The collector actor is pinned to the Ray head (num_cpus=0); autoscaler idle-down killing that actor does not fail the run.

Checkpoint/resume constraints

ParallelPlan.require_inprocess() rejects features that are not safe for multi-worker execution. Checkpoint/resume is currently inprocess-only.

On Ray workers, local offset checkpoints are skipped (persist_partition_checkpoint=False): cadence saves must not mkdir host paths such as /step01 on KubeRay pods. Run-global mutable registries cannot be merged in driver finalization; model them as partial rows followed by reduce_merge.

embed_train Ray backend

Embedding training owns run-global mutable W, so it uses a mode-specific transport (processing/embed_train/engine/backends/), not partition shard mapping.

workers = 1

• one EmbedVocabActor parameter server per material group;
• pass 1 loads/merges token counts and initializes vocabulary on the driver;
• pair batches are submitted asynchronously with bounded inflight;
• non-blocking sample polling collects train metrics;
• driver requests batched snapshots and writes them through StorageSession;
• actor is killed after group finalization.

workers ≥ 2 (honest parameter server)

runtime.parallelism.workers is the W shard count:

1. start_w_shard_actors creates one actor per shard (row_id % n_shards);
2. each train batch gathers needed rows, runs sgns_ps_step locally, pushes deltas (ray_w_shards.py);
3. snapshot merges shard exports and writes through StorageSession;
4. optional host geometry runs on the merged table (skipped for Qdrant sinks).

Backend log label is ray_ps. Step03 defaults use ray / workers: 8.

The shared run_train_over_groups controls group order/result shape for inprocess, processes, and Ray backends. Production embed_train currently hard-requires compute.embedding_method: sgns_v1.

See embed_train.

Relevant tuning params

Common execution params:

• runtime.parallelism.workers — partition shard workers or embed_train W shards on Ray
• runtime.parallelism.ray_address
• runtime.parallelism.num_gpus
• runtime.parallelism.cpus_per_worker / max_nodes — KubeRay packing (see above)
• runtime.parallelism.require_object_store
• runtime.parallelism.schedule / deflate_idle — pull queue and idle shrink (partition_loop)
• runtime.streaming.* — pending / observation / metrics queue caps
• compute.embed_ray_max_inflight — Ray train inflight batches

Embed transport params also include pair batch size, snapshot batch rows, vocabulary threshold, and geometry sampling sizes. Declare them in DRTML; do not hide them in step code.

Serialization rules

Safe worker payloads contain:

• manifest path;
• run/step/artifacts identity;
• merged params;
• shard assignment;
• simple DTOs.

Do not serialize open files, DB clients, StorageSession, metrics clients, or arbitrary domain closures. Reconstruct resources inside the worker.

Adding a Ray capability

• Connection/init / status / preflight / live RayCluster patch → integration/ray (cluster_shape.py packing, cluster_tune.py kubectl).
• Partition dispatch → processing/partition_loop/parallel/backends.
• Mode-specific transport → that mode’s backends/ package.
• Pure computation → algorithms/.
• Object-store deploy presets → integration/cloud.

Implement the same transport-neutral callable contract as existing backends, reuse shared outer loops, add bounded backpressure, explicit cleanup, failure propagation, and tests with Ray optional/skipped when unavailable.