That is the primary in a two-part collection on distributed computing utilizing Ray. This half reveals the right way to use Ray in your native PC, and half 2 reveals the right way to scale Ray to multi-server clusters within the cloud.
gotten a brand new 16-core laptop computer or desktop, and also you’re keen to check its energy with some heavy computations.
You’re a Python programmer, although not an skilled but, so that you open your favorite LLM and ask it one thing like this.
“I want to depend the variety of prime numbers inside a given enter vary. Please give me some Python code for this.”
After a couple of seconds, the LLM provides you some code. You would possibly tweak it a bit by a brief back-and-forth, and finally you find yourself with one thing like this:
import math, time, os
def is_prime(n: int) -> bool:
if n < 2: return False
if n == 2: return True
if n % 2 == 0: return False
r = int(math.isqrt(n)) + 1
for i in vary(3, r, 2):
if n % i == 0:
return False
return True
def count_primes(a: int, b: int) -> int:
c = 0
for n in vary(a, b):
if is_prime(n):
c += 1
return c
if __name__ == "__main__":
A, B = 10_000_000, 20_000_000
total_cpus = os.cpu_count() or 1
# Begin "chunky"; we will sweep this later
chunks = max(4, total_cpus * 2)
step = (B - A) // chunks
print(f"CPUs~{total_cpus}, chunks={chunks}")
t0 = time.time()
outcomes = []
for i in vary(chunks):
s = A + i * step
e = s + step if i < chunks - 1 else B
outcomes.append(count_primes(s, e))
whole = sum(outcomes)
print(f"whole={whole}, time={time.time() - t0:.2f}s")You run this system and it really works completely. The one downside is that it takes fairly a little bit of time to run , perhaps thirty to sixty seconds, relying on the scale of your enter vary. That’s most likely unacceptable.
What do you do now? You could have a number of choices, with the three commonest most likely being:
– Parallelise the code utilizing threads or multi-processing
– Rewrite the code in a “quick” language like C or Rust
– Attempt a library like Cython, Numba or NumPy
These are all viable choices, however every has disadvantages. Choices 1 and three considerably improve your code complexity, and the center choice might require you to be taught a brand new programming language.
What if I instructed you that there was one other method? One the place the adjustments required to your current code could be saved to an absolute minimal. One the place your runtime is mechanically unfold throughout all of your out there cores.
That’s exactly what the third-party Ray library guarantees to do.
What’s Ray?
The Ray Python library is an open-source distributed computing framework designed to make it simple to scale Python packages from a laptop computer to a cluster with minimal code adjustments.
Ray makes it easy to scale and distribute compute-intensive utility workloads — from deep studying to information processing — throughout clusters of distant computer systems, whereas additionally delivering sensible utility runtime enhancements in your laptop computer, desktop, or perhaps a distant cloud-based compute cluster.
Ray gives a wealthy set of libraries and integrations constructed on a versatile distributed execution framework, making distributed computing simple and accessible to all.
Briefly, Ray enables you to parallelise and distribute your Python code with minimal effort, whether or not it’s working regionally on a laptop computer or on a large cloud-based cluster.
Utilizing Ray
In the remainder of this text, I’ll take you thru the fundamentals of utilizing Ray to hurry up CPU-intensive Python code, and we’ll arrange some instance code snippets to indicate you ways simple it’s to include the ability of Ray into your individual workloads.
To get probably the most out of utilizing Ray, if you’re an information scientist or machine studying engineer, there are a couple of key ideas you must perceive first. Ray is made up of a number of elements.
Ray Knowledge is a scalable library designed for information processing in ML and AI duties. It presents versatile, high-performance APIs for AI duties, together with batch inference, information preprocessing, and information ingestion for ML coaching.
Ray Prepare is a versatile, scalable library designed for distributed machine studying coaching and fine-tuning.
Ray Tune is used for Hyperparameter Tuning.
Ray Serve is a scalable library for deploying fashions to facilitate on-line inference APIs.
Ray RLlib is used for scalable reinforcement studying
As you’ll be able to see, Ray may be very targeted on giant language fashions and AI purposes, however there’s one final essential element I haven’t talked about but, and it’s the one I’ll be utilizing on this article.
Ray Core is designed for scaling CPU-intensive general-purpose Python purposes. It’s designed to unfold your Python workload over all out there cores on whichever system you’re working it on.
This text will likely be speaking completely about Ray Core.
Two important ideas to understand inside Ray Core are duties and actors.
Duties are stateless staff or providers applied utilizing Ray by adorning common Python capabilities.
Actors (or stateful staff) are used, for instance, when you must hold monitor of and preserve the state of dependent variables throughout your distributed cluster. Actors are applied by adorning common Python lessons.
Each actors and duties are outlined utilizing the identical @ray.distant decorator. As soon as outlined, these duties are executed with the particular .distant() technique offered by Ray. We’ll have a look at an instance of this subsequent.
Organising a growth surroundings
Earlier than we begin coding, we should always arrange a growth surroundings to maintain our tasks siloed so that they don’t intervene with one another. I’ll be utilizing conda for this, however be happy to make use of whichever device you like. I’ll be working my code utilizing a Jupyter pocket book in a WSL2 Ubuntu shell on Home windows.
$ conda create -n ray-test python=3.13 -y
$ conda activate ray-test
(ray-test) $ conda set up ray[default]Code instance – counting prime numbers
Let’s revisit the instance I gave firstly: counting the variety of primes throughout the interval 10,000,000 to twenty,000,000.
We’ll run our unique Python code and time how lengthy it takes.
import math, time, os
def is_prime(n: int) -> bool:
if n < 2: return False
if n == 2: return True
if n % 2 == 0: return False
r = int(math.isqrt(n)) + 1
for i in vary(3, r, 2):
if n % i == 0:
return False
return True
def count_primes(a: int, b: int) -> int:
c = 0
for n in vary(a, b):
if is_prime(n):
c += 1
return c
if __name__ == "__main__":
A, B = 10_000_000, 20_000_000
total_cpus = os.cpu_count() or 1
# Begin "chunky"; we will sweep this later
chunks = max(4, total_cpus * 2)
step = (B - A) // chunks
print(f"CPUs~{total_cpus}, chunks={chunks}")
t0 = time.time()
outcomes = []
for i in vary(chunks):
s = A + i * step
e = s + step if i < chunks - 1 else B
outcomes.append(count_primes(s, e))
whole = sum(outcomes)
print(f"whole={whole}, time={time.time() - t0:.2f}s")And the output?
CPUs~32, chunks=64
whole=606028, time=31.17sNow, can we enhance that utilizing Ray? Sure, by following this straightforward 4-step course of.
Step 1 - Initialise Ray. Add these two strains initially of your code.
import ray
ray.init()Step 2 - Create our distant perform. That’s simple. Simply beautify the perform we need to optimise with the @ray.distant decorator. The perform to be adorned is the one which’s performing probably the most work. In our instance, that’s the count_primes perform.
@ray.distant(num_cpus=1)
def count_primes(begin: int, finish: int) -> int:
...
...Step 3 - Launch the parallel duties. Name your distant perform utilizing the .distant Ray directive.
refs.append(count_primes.distant(s, e))Step 4 - Look ahead to all our duties to finish. Every activity in Ray returns an ObjectRef when it’s been known as. It is a promise from Ray. It means Ray has set the duty off working remotely, and Ray will return its worth in some unspecified time in the future sooner or later. We monitor all of the ObjectRefs returned by working duties utilizing the ray.get() perform. This blocks till all duties have accomplished.
outcomes = ray.get(duties)Let’s put this all collectively. As you will notice, the adjustments to our unique code are minimal — simply 4 strains of code added and a print assertion to show the variety of nodes and cores we’re working on.
import math
import time
# -----------------------------------------
# Change No. 1
# -----------------------------------------
import ray
ray.init(auto)
def is_prime(n: int) -> bool:
if n < 2: return False
if n == 2: return True
if n % 2 == 0: return False
r = int(math.isqrt(n)) + 1
for i in vary(3, r, 2):
if n % i == 0:
return False
return True
# -----------------------------------------
# Change No. 2
# -----------------------------------------
@ray.distant(num_cpus=1) # pure-Python loop → 1 CPU per activity
def count_primes(a: int, b: int) -> int:
c = 0
for n in vary(a, b):
if is_prime(n):
c += 1
return c
if __name__ == "__main__":
A, B = 10_000_000, 60_000_000
total_cpus = int(ray.cluster_resources().get("CPU", 1))
# Begin "chunky"; we will sweep this later
chunks = max(4, total_cpus * 2)
step = (B - A) // chunks
print(f"nodes={len(ray.nodes())}, CPUs~{total_cpus}, chunks={chunks}")
t0 = time.time()
refs = []
for i in vary(chunks):
s = A + i * step
e = s + step if i < chunks - 1 else B
# -----------------------------------------
# Change No. 3
# -----------------------------------------
refs.append(count_primes.distant(s, e))
# -----------------------------------------
# Change No. 4
# -----------------------------------------
whole = sum(ray.get(refs))
print(f"whole={whole}, time={time.time() - t0:.2f}s")Now, has all of it been worthwhile? Let’s run the brand new code and see what we get.
2025-11-01 13:36:30,650 INFO employee.py:2004 -- Began a neighborhood Ray occasion. View the dashboard at 127.0.0.1:8265
/residence/tom/.native/lib/python3.10/site-packages/ray/_private/employee.py:2052: FutureWarning: Tip: In future variations of Ray, Ray will not override accelerator seen gadgets env var if num_gpus=0 or num_gpus=None (default). To allow this conduct and switch off this error message, set RAY_ACCEL_ENV_VAR_OVERRIDE_ON_ZERO=0
warnings.warn(
nodes=1, CPUs~32, chunks=64
whole=606028, time=3.04sProperly, the consequence speaks for itself. The Ray Python code is 10x quicker than the common Python code. Not too shabby.
The place does this improve in pace come from? Properly, Ray can unfold your workload to all of the cores in your system. A core is sort of a mini-CPU. After we ran our unique Python code, it used just one core. That’s high quality, but when your CPU has multiple core, which most trendy PCs do, then you definitely’re leaving cash on the desk, so to talk.
In my case, the CPU has 24 cores, so it’s not shocking that my Ray code was method quicker than the non-Ray code.
Monitoring Ray jobs
One other level price mentioning is that Ray makes it very simple to observe job executions by way of a dashboard. Discover within the output we obtained when working our Ray instance code, we noticed this,
... -- Began a neighborhood Ray occasion. View the dashboard at 127.0.0.1:8265It’s exhibiting a neighborhood URL hyperlink as a result of I’m working this on my desktop. If you happen to had been working this on a cluster, the URL would level to a location on the cluster head node.
Once you click on on the given URL hyperlink, it’s best to see one thing just like this,
From this primary display screen, you’ll be able to drill down to observe many points of your Ray packages utilizing the menu hyperlinks on the high of the web page.
Utilizing Ray actors
I beforehand talked about that actors had been an integral a part of the Ray core processing. Actors are used to coordinate and share information between Ray duties. For instance, say you need to set a world restrict for ALL working duties that they need to adhere to. Let’s say you will have a pool of employee duties, however you need to make sure that solely a most of 5 of these duties can run concurrently. Right here is a few code you would possibly suppose would work.
import math, time, os
def is_prime(n: int) -> bool:
if n < 2: return False
if n == 2: return True
if n % 2 == 0: return False
r = int(math.isqrt(n)) + 1
for i in vary(3, r, 2):
if n % i == 0:
return False
return True
def count_primes(a: int, b: int) -> int:
c = 0
for n in vary(a, b):
if is_prime(n):
c += 1
return c
if __name__ == "__main__":
A, B = 10_000_000, 20_000_000
total_cpus = os.cpu_count() or 1
# Begin "chunky"; we will sweep this later
chunks = max(4, total_cpus * 2)
step = (B - A) // chunks
print(f"CPUs~{total_cpus}, chunks={chunks}")
t0 = time.time()
outcomes = []
for i in vary(chunks):
s = A + i * step
e = s + step if i < chunks - 1 else B
outcomes.append(count_primes(s, e))
whole = sum(outcomes)
print(f"whole={whole}, time={time.time() - t0:.2f}s")We now have used a world variable to restrict the variety of working duties, and the code is syntactically appropriate, working with out error. Sadly, you received’t get the consequence you anticipated. That’s as a result of every Ray activity runs in its personal course of area and has its personal copy of the worldwide variable. The worldwide variable is NOT shared between capabilities. So once we run the above code, we’ll see output like this,
Complete calls: 200
Meant GLOBAL_QPS: 5.0
Anticipated time if really global-limited: ~40.00s
Precise time with 'world var' (damaged): 3.80s
Noticed cluster QPS: ~52.6 (ought to have been ~5.0)To repair this, we use an actor. Recall that an actor is only a Ray-decorated Python class. Right here is the code with actors.
import time, ray
ray.init(ignore_reinit_error=True, log_to_driver=False)
# That is our actor
@ray.distant
class GlobalPacer:
"""Serialize calls so cluster-wide price <= qps."""
def __init__(self, qps: float):
self.interval = 1.0 / qps
self.next_time = time.time()
def purchase(self):
# Wait contained in the actor till we will proceed
now = time.time()
if now < self.next_time:
time.sleep(self.next_time - now)
# Reserve the following slot; guard in opposition to drift
self.next_time = max(self.next_time + self.interval, time.time())
return True
@ray.distant
def call_api_with_limit(n_calls: int, pacer):
performed = 0
for _ in vary(n_calls):
# Look ahead to world permission
ray.get(pacer.purchase.distant())
# fake API name (no further sleep right here)
performed += 1
return performed
if __name__ == "__main__":
NUM_WORKERS = 10
CALLS_EACH = 20
GLOBAL_QPS = 5.0 # cluster-wide cap
total_calls = NUM_WORKERS * CALLS_EACH
expected_min_time = total_calls / GLOBAL_QPS
pacer = GlobalPacer.distant(GLOBAL_QPS)
t0 = time.time()
ray.get([call_api_with_limit.remote(CALLS_EACH, pacer) for _ in range(NUM_WORKERS)])
dt = time.time() - t0
print(f"Complete calls: {total_calls}")
print(f"International QPS cap: {GLOBAL_QPS}")
print(f"Anticipated time (if capped at {GLOBAL_QPS} QPS): ~{expected_min_time:.2f}s")
print(f"Precise time with actor: {dt:.2f}s")
print(f"Noticed cluster QPS: ~{total_calls/dt:.1f}")Our limiter code is encapsulated in a category (GlobalPacer) and adorned with ray.distant, which means it applies to all working duties. We are able to see the distinction this makes to the output by working the up to date code.
Complete calls: 200
International QPS cap: 5.0
Anticipated time (if capped at 5.0 QPS): ~40.00s
Precise time with actor: 39.86s
Noticed cluster QPS: ~5.0Abstract
This text launched Ray, an open-source Python framework that makes it simple to scale compute-intensive packages from a single core to a number of cores or perhaps a cluster with minimal code adjustments.
I briefly talked about the important thing elements of Ray—Ray Knowledge, Ray Prepare, Ray Tune, Ray Serve, and Ray Core—emphasising that Ray Core is right for general-purpose CPU scaling.
I defined a few of the important ideas in Ray Core, corresponding to its introduction of duties (stateless parallel capabilities), actors (stateful staff for shared state and coordination), and ObjectRefs (a future promise of a activity’s return worth)
To showcase some great benefits of utilizing Ray, I started with a easy CPU-intensive instance — counting prime numbers over a variety — and confirmed how working it on a single core will be sluggish with a naive Python implementation.
As a substitute of rewriting the code in one other language or utilizing complicated multiprocessing libraries, Ray lets you parallelise the workload in simply 4 easy steps and only a few further strains of code:
- ray.init() to begin Ray
- Embellish your capabilities with @ray.distant to show them into parallel duties
- .distant() to launch duties concurrently, and
- ray.get() to gather activity outcomes.
This method minimize the runtime of the prime-counting instance from ~30 seconds to ~3 seconds on a 24-core machine.
I additionally talked about how simple it’s to observe working jobs in Ray utilizing its built-in dashboard and confirmed the right way to entry it.
Lastly, I offered an instance of utilizing a Ray Actor by exhibiting why world variables usually are not appropriate for coordinating throughout duties, since every employee has its personal reminiscence area.
Within the second a part of this collection, we’ll see the right way to take issues to a different degree by enabling Ray jobs to make use of much more CPU energy as we scale to giant multi-node servers within the cloud by way of Amazon Net Companies.
