BUG: Modin with Ray not scaling across multiple CPUs

[simone-macri]

Modin version checks

• I have checked that this issue has not already been reported.

• I have confirmed this bug exists on the latest released version of Modin.

• I have confirmed this bug exists on the main branch of Modin. (In order to do this you can follow this guide.)

Reproducible Example

import numpy as np
from modin.config import Engine, CpuCount, NPartitions
import ray
import modin.pandas as pd
import os

# Set environment variables before importing Modin
os.environ["MODIN_ENGINE"] = "ray"
os.environ["MODIN_CPUS"] = "11"

# Initialize Ray with the desired number of CPUs
ray.init(num_cpus=11)

# Optionally, set Modin configuration
Engine.put("ray")
CpuCount.put(11)

# Confirm Ray resources
print(ray.cluster_resources())
print("Number of partitions:", NPartitions.get())

# Create a large DataFrame and perform operations
df = pd.DataFrame(np.random.rand(10**7, 10))
result = df.apply(lambda x: x**2)

Issue Description

I am experiencing an issue where Modin with the Ray backend does not scale across multiple CPUs. Despite configuring the environment to use 11 CPUs (MODIN_ENGINE=ray, MODIN_CPUS=11) and initializing Ray with ray.init(num_cpus=11), there is no noticeable improvement in execution times compared to using just 1 CPU.

The problem persists even with large computational workloads, such as applying functions over DataFrames with millions of rows. Monitoring CPU usage during the operation shows that all CPUs are not being fully utilized.

This behavior suggests that either the parallelism is not correctly distributed or there is some bottleneck preventing Modin from utilizing the available resources effectively.

Expected Behavior

When configuring Modin with the Ray backend to use 11 CPUs, I expect the execution times for operations on large DataFrames to improve significantly compared to using a single CPU. Specifically:

1. Parallelism: All 11 CPUs should be utilized effectively during computationally intensive operations, such as applying functions across rows or columns in a large DataFrame.
2. Performance Improvement: The execution time should decrease as the number of CPUs increases, reflecting efficient parallel processing.
3. Resource Utilization: Monitoring tools like htop or the Ray dashboard should show all CPUs actively working during the operation.

Error Logs

No response

Installed Versions

INSTALLED VERSIONS

commit : 3e951a6
python : 3.12.7
python-bits : 64
OS : Linux
OS-release : 6.10.14-linuxkit
Version : #1 SMP Thu Oct 24 19:28:55 UTC 2024
machine : aarch64
processor :
byteorder : little
LC_ALL : None
LANG : C.UTF-8
LOCALE : C.UTF-8

Modin dependencies

modin : 0.32.0
ray : 2.39.0
dask : 2024.11.2
distributed : 2024.11.2

pandas dependencies

pandas : 2.2.3
numpy : 2.1.3
pytz : 2024.2
dateutil : 2.9.0.post0
pip : 24.2
Cython : None
sphinx : None
IPython : 8.29.0
adbc-driver-postgresql: None
adbc-driver-sqlite : None
bs4 : 4.12.3
blosc : None
bottleneck : None
dataframe-api-compat : None
fastparquet : None
fsspec : 2024.10.0
html5lib : None
hypothesis : None
gcsfs : None
jinja2 : 3.1.4
lxml.etree : None
matplotlib : None
numba : None
numexpr : None
odfpy : None
openpyxl : None
pandas_gbq : None
psycopg2 : None
pymysql : None
pyarrow : 18.0.0
pyreadstat : None
pytest : None
python-calamine : None
pyxlsb : None
s3fs : None
scipy : None
sqlalchemy : None
tables : None
tabulate : None
xarray : None
xlrd : None
xlsxwriter : None
zstandard : None
tzdata : 2024.2
qtpy : None
pyqt5 : None