The Ray object store is a distributed, in-memory key-value store that Ray uses to manage data shared between tasks and actors.
Let’s see it in action. Imagine you have a large NumPy array that you want to share between multiple Ray tasks.
import ray
import numpy as np
ray.init()
# Create a large NumPy array
data = np.random.rand(10000, 10000)
# Put the array into the Ray object store
object_ref = ray.put(data)
# Now, multiple tasks can access this data efficiently
@ray.remote
def process_data(data_ref):
# When the task starts, Ray fetches the data from the object store
# and makes it available as a local NumPy array.
local_data = ray.get(data_ref)
return np.sum(local_data)
# Launch two tasks that will process the same data
result1_ref = process_data.remote(object_ref)
result2_ref = process_data.remote(object_ref)
# Get the results
result1 = ray.get(result1_ref)
result2 = ray.get(result2_ref)
print(f"Result 1: {result1}")
print(f"Result 2: {result2}")
ray.shutdown()
In this example, ray.put(data) doesn’t copy the data. Instead, it stores the NumPy array in the Ray object store and returns an ObjectRef. This ObjectRef is a lightweight handle. When process_data.remote(object_ref) is called, Ray doesn’t serialize and send the entire NumPy array to the worker process. Instead, it sends the ObjectRef. The worker process then uses this ObjectRef to fetch the actual data from the object store. If the data is already local to the worker, it’s accessed directly. If not, Ray efficiently transfers it from the node where it resides.
The core problem the Ray object store solves is efficient data sharing in a distributed environment. Naively sending large Python objects (like NumPy arrays, Pandas DataFrames, or custom objects) between processes or machines involves serialization (converting the object to a byte stream) and deserialization. This is slow and memory-intensive. The object store acts as a central, distributed memory that Ray processes can access directly. When you ray.put an object, Ray serializes it once and stores it in the object store. Subsequent accesses, whether by different tasks on the same node or tasks on different nodes, involve fetching the already-serialized bytes and deserializing them only when needed. This significantly reduces data transfer overhead and memory duplication.
Internally, the object store is a distributed hash table. Each worker process has a local memory manager that interacts with the global object store. When you call ray.put(obj), the object is serialized and sent to a designated object store server (or multiple servers for redundancy). A unique Object ID is generated and associated with this serialized data. This Object ID is what gets returned as the ObjectRef. When another task calls ray.get(object_ref), the Object ID is used to look up the data in the object store. The data is then transferred to the requesting process and deserialized. Ray employs a sophisticated scheduling mechanism to ensure data is fetched efficiently, prioritizing local copies and minimizing network traffic.
The size of the object store is a crucial tuning parameter. By default, Ray might not allocate a huge amount of memory for the object store, especially if you’re running locally. If you’re dealing with very large datasets that need to be shared, you’ll want to explicitly configure the object store size. This is typically done when you initialize Ray:
ray.init(object_store_memory=10e9) # Allocate 10 GB for the object store
This command tells Ray to reserve 10 gigabytes of memory specifically for the object store on each node. Without sufficient memory, Ray might evict objects from the store prematurely, forcing recomputation or re-transfer, which negates the performance benefits.
A common misconception is that ray.put immediately makes the data available everywhere. In reality, ray.put serializes and stores the object, returning a reference. The actual data transfer to remote nodes happens lazily, only when a ray.get operation for that ObjectRef is initiated on a node that doesn’t have a local copy. This lazy fetching is key to Ray’s efficiency, as data is only moved when it’s actually needed.
The default behavior for object eviction from the object store when memory is full is least-recently-used (LRU). You can influence this behavior and understand memory usage with ray.experimental.get_object_store_memory_usage().
The next conceptual hurdle is understanding how Ray actors leverage the object store for their internal state and method calls.