Scaling Python Workloads with Ray on Kubernetes

01.08.2025 — ray, kubernetes, ml, kuberay — 8 min read

Python is the language of machine learning, but the Global Interpreter Lock (GIL) and a single-node design cap how far a plain Python program scales. Ray is a distributed runtime that lets you scale ordinary Python from a laptop to a cluster with small code changes, and KubeRay runs it on Kubernetes.

The whole model rests on three primitives: remote tasks (stateless functions), actors (stateful workers), and a shared object store for zero-copy data. On Kubernetes you run them through the KubeRay operator and three CRDs (custom resource definitions) in the ray.io/v1 API group: RayCluster, RayJob, and RayService. Two details decide whether your first cluster comes up. A valid RayCluster needs rayVersion, a head pod template, and worker groups with groupName, replicas, and a pod template; minimal examples that omit the template will not apply. And autoscaling is opt-in, with enableInTreeAutoscaling: true plus minReplicas/maxReplicas on worker groups, not just editing replicas. This post walks from that core model up to a running, autoscaling cluster.

Where other frameworks stop

Other distributed frameworks each miss part of the modern ML workload. MPI gives fine-grained, low-latency control but is complex and tied to static HPC setups. Apache Spark is excellent for batch data processing but awkward for dynamic, GPU-heavy, actor-style work. Dask brings Pythonic parallelism but struggles with long-lived stateful actors and model serving.

Ray targets that gap: a Python-first runtime for dynamic task graphs, stateful actors, and model serving, scaling from a laptop to thousands of nodes. You write normal Python and annotate the parts that should run remotely.

Tasks, actors, and a shared store

Start with the smallest piece. Install Ray and try a remote task. @ray.remote turns a function into a task you invoke with .remote(), which returns a future you resolve with ray.get().

import ray
ray.init()

@ray.remote
def square(x):
    return x * x

futures = [square.remote(i) for i in range(5)]
print(ray.get(futures))  # [0, 1, 4, 9, 16]

Notice .remote() schedules the work and returns immediately; ray.get() blocks until the results are ready. This is the whole task API.

The object store

Task results live in Ray's in-memory object store, based on shared memory (Plasma). Objects are shared between workers on the same node without copying, which avoids the serialization cost that dominates naive distributed Python. A future (ObjectRef) is a handle to a value in that store.

Actors: stateful workers

When you need to keep state, like a counter or a loaded model, use an actor: a class annotated with @ray.remote. Each actor is a long-lived process, and method calls are remote.

@ray.remote
class Counter:
    def __init__(self):
        self.count = 0
    def increment(self):
        self.count += 1
        return self.count

c = Counter.remote()
print(ray.get(c.increment.remote()))  # 1
print(ray.get(c.increment.remote()))  # 2

Notice the actor holds self.count across calls, which a task cannot. Actors are how you model services, simulations, and stateful ML components.

Task graphs

Passing one task's future into another builds a dependency graph. Ray schedules each task when its inputs are ready.

@ray.remote
def add(x, y):
    return x + y

a = add.remote(1, 2)
b = add.remote(3, 4)
c = add.remote(a, b)   # depends on a and b
print(ray.get(c))      # 10

Notice you pass a and b (futures), not resolved values. Ray sees the dependency and runs add(a, b) only after both complete, maximizing parallelism.

How a Ray cluster is wired

Those primitives run on a specific topology. A Ray cluster has a head node and worker nodes. Each node runs a raylet, which manages that node's resources and schedules tasks locally. The head node additionally runs the Global Control Store (GCS), which holds cluster metadata. Scheduling is decentralized: raylets cooperate rather than routing everything through a single global scheduler.

Head node
      ┌────────────────────────────┐
      │ GCS (cluster metadata)      │
      │ raylet + object store       │
      └────────────────────────────┘
          │                    │
   Worker node            Worker node
   raylet + store         raylet + store
   tasks / actors         tasks / actors

Caption: the head node holds the GCS; every node runs a raylet and a slice of the object store.

Fault tolerance comes from lineage: if a node fails, Ray can reconstruct lost objects by re-running the tasks that produced them, using the recorded dependency graph.

Getting Ray onto Kubernetes

With the shape of a cluster clear, put one on Kubernetes. KubeRay is the operator that manages Ray on Kubernetes. Install it with Helm, pinning a version so your clusters are reproducible. This installs the CRDs and the operator into the default namespace.

helm repo add kuberay https://ray-project.github.io/kuberay-helm/
helm install kuberay-operator kuberay/kuberay-operator --version 1.4.2
kubectl get pods   # kuberay-operator pod should be Running

Notice the release is kuberay-operator and the --version flag pins the operator (and CRDs) to a known release. For local testing, create a cluster first with kind create cluster.

A correct RayCluster

A RayCluster needs rayVersion (matching the image), a head pod template, and one or more worker groups. Each worker group needs a groupName, replica counts, and its own pod template. Set CPU and memory on the container, not only in rayStartParams.

apiVersion: ray.io/v1
kind: RayCluster
metadata:
  name: raycluster
spec:
  rayVersion: "2.46.0"
  headGroupSpec:
    rayStartParams: {}
    template:
      spec:
        containers:
          - name: ray-head
            image: rayproject/ray:2.46.0
            resources:
              requests: { cpu: "1", memory: 2Gi }
              limits: { cpu: "1", memory: 2Gi }
  workerGroupSpecs:
    - groupName: small-group
      replicas: 2
      minReplicas: 2
      maxReplicas: 5
      rayStartParams: {}
      template:
        spec:
          containers:
            - name: ray-worker
              image: rayproject/ray:2.46.0
              resources:
                requests: { cpu: "2", memory: 4Gi }
                limits: { cpu: "2", memory: 4Gi }

Notice the worker image must match rayVersion, and resources on the container is what actually reserves CPU and memory. rayStartParams: {} is required even when empty. Apply with kubectl apply -f raycluster.yaml and watch for head and worker pods.

The Ray dashboard

Forward the head service to reach the dashboard, which shows tasks, actors, and resource usage.

kubectl port-forward svc/raycluster-head-svc 8265:8265

Notice the service is <cluster-name>-head-svc. Open http://localhost:8265.

Submitting a batch job

A long-running cluster is one use; a job that spins up, runs, and cleans up is another. For a batch job, use the RayJob CRD (also ray.io/v1). It can create a cluster for the job with rayClusterSpec, or run on an existing cluster with clusterSelector. entrypoint is the command KubeRay submits, and shutdownAfterJobFinishes tears the cluster down when done.

apiVersion: ray.io/v1
kind: RayJob
metadata:
  name: example-rayjob
spec:
  entrypoint: python /home/ray/samples/sample_code.py
  shutdownAfterJobFinishes: true
  clusterSelector:
    ray.io/cluster: raycluster   # run on the existing RayCluster above

Notice entrypoint is a normal shell command. With clusterSelector the job reuses your cluster; drop it and add rayClusterSpec to have KubeRay create a dedicated cluster per job. Submit with kubectl apply -f rayjob.yaml.

Letting the cluster grow itself

Manually editing replicas is not autoscaling. Ray's autoscaler watches pending tasks and actors and scales worker groups between their bounds. Turn it on with enableInTreeAutoscaling: true, which adds an autoscaler sidecar to the head pod, and set minReplicas/maxReplicas per worker group.

spec:
  enableInTreeAutoscaling: true
  workerGroupSpecs:
    - groupName: gpu-group
      minReplicas: 0
      maxReplicas: 8
      rayStartParams: {}
      template:
        spec:
          containers:
            - name: ray-worker
              image: rayproject/ray:2.46.0
              resources:
                limits: { cpu: "4", memory: 16Gi, nvidia.com/gpu: "1" }

Notice minReplicas: 0 lets an expensive GPU group scale to zero when idle, and the autoscaler adds workers only when tasks request GPUs. Request GPUs through the container resources.limits.

Running your code against the cluster

The recommended way to run code against the cluster is the Ray Jobs API. Port-forward the dashboard and submit:

kubectl port-forward svc/raycluster-head-svc 8265:8265
ray job submit --address http://localhost:8265 -- python my_script.py

Notice ray job submit packages your script and runs it on the cluster, streaming logs back. Inside a pod on the cluster, ray.init(address="auto") connects to the local Ray instance directly.

What builds on top

Ray's libraries build on the same cluster: Ray Tune for hyperparameter search, Ray Serve for model serving (RayService CRD on Kubernetes), RLlib for reinforcement learning, and Ray Data for distributed loading and preprocessing. They share the cluster's scheduler and object store, so you can chain data loading, training, and serving without leaving Ray.

Pitfalls that bite

Most of the trouble people hit clusters around a few recurring mistakes.

RayCluster will not apply. A spec missing rayVersion, the head template, or a worker groupName is invalid. Use the full shape above, not a stripped-down snippet.
Version mismatch. The container image tag must match rayVersion. A mismatch causes obscure runtime failures.
CPU/GPU not honored. Ray reserves resources from the container resources, not from rayStartParams alone. Set requests/limits.
Autoscaler does nothing. You set maxReplicas but forgot enableInTreeAutoscaling: true, so no autoscaler sidecar exists.
Old ray.io/v1alpha1 YAML. RayJob and RayService are ray.io/v1 now. Copying a 2022-era v1alpha1 RayJob will fail to validate.
Client connection. Prefer ray job submit over the older Ray Client (ray://) address for running code from outside the cluster.

From a laptop script to a GPU cluster

Ray scales Python with a small, consistent model: tasks, actors, and a shared object store. On Kubernetes, the KubeRay operator and the ray.io/v1 CRDs (RayCluster, RayJob, RayService) run it in production. The two things people get wrong are the RayCluster shape (it needs rayVersion and pod templates) and autoscaling (it needs enableInTreeAutoscaling plus replica bounds). Get those right and you can go from a laptop script to an autoscaling GPU cluster with the same code.