Ray

Ray Core executes Python functions and classes as distributed tasks across a cluster using a Raylet-based architecture where each node runs a Raylet daemon that manages local task scheduling and execution. Tasks are submitted to a Global Control Store (GCS) which coordinates scheduling across nodes, while an object store (Apache Arrow-based) handles inter-task data transfer with zero-copy semantics. The system uses compiled DAGs for accelerated execution paths that bypass the task submission overhead for tightly-coupled workloads.

Ray Train (v2) abstracts distributed training orchestration through a controller-worker architecture where a central controller coordinates training across worker groups, handling data loading, checkpoint management, and fault tolerance. It integrates natively with PyTorch, TensorFlow, Hugging Face Transformers, and DeepSpeed via framework-specific adapters that inject Ray's distributed primitives (data sharding, gradient synchronization) without modifying user training code. Runtime environments ensure consistent dependency versions across workers via containerization or conda environment replication.

Ray's compiled DAG feature compiles static task graphs into optimized execution plans that bypass the task submission queue, reducing per-task overhead from ~5-10ms to <1ms. DAGs are defined using ray.dag API where tasks are connected as a directed acyclic graph, then compiled into a single execution unit. Compiled DAGs execute entirely on the cluster without returning to the client, enabling tight loops of dependent tasks with minimal latency. This is particularly useful for serving pipelines where requests flow through multiple model inference stages.

Ray's object store uses Apache Arrow for efficient in-memory data representation, enabling zero-copy data transfer between tasks on different nodes via shared memory or network protocols. Objects are stored in a distributed object store where each node maintains a local store, and the GCS tracks object locations. When a task needs an object on a remote node, Ray uses efficient transfer protocols (RDMA when available, TCP fallback) to move data without serialization overhead. Large objects are automatically spilled to disk when memory is exhausted, with configurable spilling policies.

Ray Tune executes hyperparameter search by spawning trial actors that run training code in parallel, coordinating via a central trial manager that tracks metrics and applies search algorithms (grid search, random search, Bayesian optimization, population-based training). Early stopping schedulers (ASHA, Median Stopping Rule) evaluate trial progress at regular intervals and terminate unpromising trials, reallocating resources to better-performing configurations. Search algorithms receive trial results via a callback interface and suggest new hyperparameters, enabling adaptive search strategies that exploit intermediate results.

Ray Data provides a distributed DataFrame-like API that executes transformations (map, filter, groupby, join) as lazy task graphs compiled into execution plans. Data is partitioned across cluster nodes and processed in streaming fashion where possible, with automatic resource management that balances memory usage and throughput. Sources (Parquet, CSV, S3, databases) and sinks (Parquet, Delta, databases) are abstracted via pluggable connectors that handle distributed I/O. For LLM workloads, Ray Data includes specialized operators for tokenization, embedding, and batch inference that integrate with Hugging Face and vLLM.

Ray Serve deploys models as stateless or stateful deployment actors that receive HTTP/gRPC requests routed through a load balancer. Deployments support dynamic batching where requests are accumulated and processed together, reducing per-request overhead for inference. Request routing uses a composable DAG where multiple deployments can be chained (e.g., preprocessing → model → postprocessing), with automatic request multiplexing and response aggregation. Ray Serve LLM provides specialized deployments for LLM serving with token streaming, prompt caching, and integration with vLLM for efficient batch inference.

Ray's autoscaler monitors cluster resource utilization and pending tasks, automatically launching new nodes when demand exceeds capacity and terminating idle nodes to reduce costs. Scheduling decisions are resource-aware: tasks specify CPU/GPU/memory requirements, and the scheduler places tasks on nodes with sufficient resources, triggering node launches if no suitable nodes exist. Node labels enable placement constraints (e.g., 'gpu_type:a100') for heterogeneous clusters. The autoscaler integrates with cloud providers (AWS, GCP, Azure) via cloud-specific drivers that handle instance launch/termination.

Ray's runtime environment system ensures consistent Python dependencies across all cluster nodes by syncing conda environments or pip packages from the client to workers. Environments can be specified per-job or per-task, enabling different jobs to use different dependency versions without conflicts. The system handles dependency resolution, caching, and installation on remote nodes, with support for custom Python paths and compiled extensions. Runtime environments are containerized via Docker when specified, enabling reproducibility across different infrastructure.

Ray provides a web-based dashboard that visualizes cluster state (nodes, actors, tasks) in real-time, showing resource utilization, task execution timeline, and error logs. The State API exposes cluster metadata (tasks, actors, jobs) as queryable objects, enabling programmatic monitoring and debugging. Metrics are exported in Prometheus format for integration with external monitoring systems (Datadog, New Relic). Distributed tracing via OpenTelemetry captures request flow across actors and tasks, enabling performance analysis of complex workloads.

Ray provides fault tolerance through automatic checkpointing of actor state and task results. When a node fails, Ray detects the failure via heartbeat timeout and reschedules affected tasks on healthy nodes. For stateful actors, checkpoints are persisted to external storage (S3, GCS, local filesystem), enabling recovery of actor state after failure. Ray Train integrates checkpointing with training loops, automatically saving model weights and optimizer state at regular intervals. The system supports both synchronous checkpointing (blocking training) and asynchronous checkpointing (background save).

Ray RLlib executes RL training by distributing environment sampling across worker actors that collect rollouts in parallel, while a learner actor trains the policy on collected data. The system supports both on-policy algorithms (PPO, A3C) that require fresh samples and off-policy algorithms (DQN, SAC) that replay stored experiences. Environment workers are stateful actors that maintain environment instances, enabling efficient sample collection without environment re-initialization. The framework abstracts algorithm implementation, allowing users to specify algorithm configuration (learning rate, network architecture) without implementing gradient updates.