Modin df becomes pandas df when sent to a ray remote. How can I stop it doing that?

[Liquidmasl]

Hi there,

I am not only using modin by itself, i also use ray quite a lot. It is now necessary to send a full modin datagrame to a ray remote Actor/Task. But whenever this is done, in the actor/task the modin dataframe is no longer a modin dataframe. It becomes a default pandas dataframe.

Even when the modin df is encapsulated in a class that is sent to a remote, in the remote its a pandas again.

How can I make it not do that? currently the only thing I managed is getting partition list with from_partition, send it to a remote, and then reconstruct the modin df there. Which is kinda annoying, and also a bit slow.

Any chance of telling modin to stay a modin df?

[NguyenTienQuoc]

Hi @Liquidmasl,

1. Root Cause

A modin.pandas.DataFrame is not preserved as a Modin DataFrame when it is passed directly across a Ray remote-task boundary in the reproduced setup. The remote task receives a regular pandas.DataFrame instead.

This is a consequence of how the Ray-backed Modin DataFrame is represented and serialized across the Ray boundary: the high-level Modin DataFrame wrapper is not reconstructed as a modin.pandas.DataFrame in the receiving worker.

Wrapping the DataFrame in another Python object does not change the result. The Modin DataFrame is still part of the object being serialized, so the worker still receives the DataFrame in its deserialized Pandas representation.

In other words, the important distinction is:

modin.pandas.DataFrame
        |
        | Ray serialization boundary
        v
pandas.DataFrame

rather than:

modin.pandas.DataFrame
        |
        | Ray serialization boundary
        v
modin.pandas.DataFrame

2. Reproducible Environment

The reproduction was run with:

Python: 3.10.11
Modin: 0.37.1
Ray: 2.57.0
Execution engine: Ray
Dataset size: 500,000 rows
Number of partitions: 16

The version information was obtained directly from the environment used for the reproduction.

3. Minimal Reproducible Example

The following reproduces the behavior with the Ray execution engine:

import os

os.environ["MODIN_ENGINE"] = "ray"

import modin
import modin.pandas as mpd
import ray

print("Python:", __import__("sys").version)
print("Modin:", modin.__version__)
print("Ray:", ray.__version__)

ray.init(ignore_reinit_error=True)


@ray.remote
def check_direct_type(df):
    return str(type(df))


class DataContainer:
    def __init__(self, df):
        self.df = df


@ray.remote
def check_container_type(container):
    return str(type(container.df))


m_df = mpd.DataFrame({
    "col_a": [1, 2, 3],
    "col_b": [4, 5, 6],
})

print("Caller type:", type(m_df))
# <class 'modin.pandas.dataframe.DataFrame'>

direct_type = ray.get(check_direct_type.remote(m_df))
print("Inside worker (direct):", direct_type)
# <class 'pandas.core.frame.DataFrame'>

container_type = ray.get(
    check_container_type.remote(DataContainer(m_df))
)
print("Inside worker (container):", container_type)
# <class 'pandas.core.frame.DataFrame'>

Thus, simply putting the Modin DataFrame inside another class does not preserve its Modin type.

---

4. Official / Source References

Modin's Ray backend represents a Modin DataFrame using Ray-backed partitions and Modin's internal dataframe/query-compiler layers.

The reconstruction approach below uses classes from modin.core, such as PandasOnRayDataframe and PandasQueryCompiler. These are implementation details rather than stable public APIs.

For applications that want to move from Modin to a Ray-native distributed data abstraction, Modin also provides a Ray-specific conversion API:

df.modin.to_ray()

This converts a Modin DataFrame/Series to a Ray Dataset when the Ray execution engine is being used.

Therefore, there are two different goals here:

1. Preserve the modin.pandas.DataFrame abstraction inside the Ray worker.

2. Move the data into a Ray-native abstraction such as ray.data.Dataset.

The first requires reconstruction of the Modin dataframe representation; the second can use Modin's Ray conversion API.

5. Why the Manual Partition Reconstruction Works

The existing workaround of sending the partitions and reconstructing the DataFrame on the worker works because it avoids sending the high-level modin.pandas.DataFrame wrapper itself.

Conceptually:

Modin DataFrame
      |
      +-- partitions
      |
      v
Ray task
      |
      v
PandasOnRayDataframe
      |
      v
PandasQueryCompiler
      |
      v
modin.pandas.DataFrame

For example:

from modin.core.execution.ray.implementations.pandas_on_ray.dataframe.dataframe import (
    PandasOnRayDataframe,
)
from modin.core.storage_formats.pandas.query_compiler import (
    PandasQueryCompiler,
)


@ray.remote
def reconstruct_worker_task(partitions):
    frame = PandasOnRayDataframe(partitions)
    query_compiler = PandasQueryCompiler(frame)
    reconstructed_df = mpd.DataFrame(query_compiler=query_compiler)

    return str(type(reconstructed_df))

This successfully reconstructs a modin.pandas.DataFrame in the worker.

However, there are two important limitations:

1. It relies on Modin internal APIs.

   PandasOnRayDataframe, PandasQueryCompiler, _query_compiler, _modin_frame, and _partitions are implementation details under modin.core rather than stable public APIs.

2. It requires explicit reconstruction.

   The worker has to rebuild the Modin DataFrame and its query compiler instead of receiving the original high-level object directly.

In my reproduction, this approach preserved the Modin type, but the total end-to-end time was approximately 4.39 seconds.

Because the reconstruction itself creates a Modin dataframe backed by Ray partitions inside a Ray task, I would treat this as an internal workaround and use caution with more complex nested Ray workloads. The benchmark below demonstrates the type preservation and timing, but does not by itself establish that a deadlock will always occur.

---

6. Workarounds and Benchmark

I tested three alternatives in addition to the direct and manual approaches.

Approach	Object received by worker	Preserves Modin DF?	Measured time	Main idea
Direct passing	pandas.DataFrame	No	~2.12 s	Default serialization behavior
Encapsulation	pandas.DataFrame	No	~0.25 s	Wrapping the DF does not change serialization
Manual reconstruction	modin.pandas.DataFrame	Yes	~4.39 s	Send partitions and rebuild Modin
Custom __reduce__	modin.pandas.DataFrame	Yes	~0.43 s	Automate worker-side reconstruction
Partition ObjectRefs	pandas.DataFrame chunks	No	~0.05 s	Process partitions directly
Ray Data	ray.data.Dataset	No	~5.33 s*	Use Ray's native data abstraction

The measurements above are from a 500,000-row reproduction with 16 partitions.

The timings are measurements from this reproduction, not general performance guarantees.

The Ray Data run also produced a Raylet Windows fatal exception: access violation message during shutdown in this particular Windows environment. The Dataset itself was successfully created and received by the worker, but I would not use this single run as evidence that Ray Data is universally safe or faster/slower than the other approaches.

7. Option 1 — Custom __reduce__ Wrapper

If the requirement is specifically:

"I want the remote task to receive a modin.pandas.DataFrame."

then a custom serialization wrapper can automate the same reconstruction that is currently being done manually.

from modin.core.execution.ray.implementations.pandas_on_ray.dataframe.dataframe import (
    PandasOnRayDataframe,
)
from modin.core.storage_formats.pandas.query_compiler import (
    PandasQueryCompiler,
)


def _reconstruct_modin_dataframe(partitions):
    frame = PandasOnRayDataframe(partitions)
    query_compiler = PandasQueryCompiler(frame)

    return mpd.DataFrame(query_compiler=query_compiler)


class ModinContainer:
    def __init__(self, modin_df):
        self.partitions = (
            modin_df
            ._query_compiler
            ._modin_frame
            ._partitions
        )

    def __reduce__(self):
        return (
            _reconstruct_modin_dataframe,
            (self.partitions,),
        )


@ray.remote
def process_modin_task(df):
    return type(df), df.shape


result_type, shape = ray.get(
    process_modin_task.remote(
        ModinContainer(m_df)
    )
)

print(result_type)
# <class 'modin.pandas.dataframe.DataFrame'>

print(shape)
# (3, 2)

The important part is that the wrapper controls how the object is reconstructed:

ModinContainer
        |
        | __reduce__()
        v
partition references + reconstruction function
        |
        v
Ray worker
        |
        v
modin.pandas.DataFrame

This avoids explicitly writing the reconstruction logic at every call site.

In my reproduction, this approach took approximately 0.43 seconds and preserved the Modin DataFrame type inside the remote task.

However, this is also an internal workaround. It depends on Modin's internal Ray backend classes and private attributes, so I would not recommend relying on it as a stable public API across Modin releases.

---

8. Option 2 — Pass Partition ObjectRefs Directly

If the actual goal is to perform parallel computation on the underlying partitions rather than to keep the Modin DataFrame abstraction, passing the partition ObjectRefs directly is considerably simpler and faster in this reproduction.

@ray.remote
def process_partition(pandas_chunk):
    return pandas_chunk.describe()


partitions = (
    m_df
    ._query_compiler
    ._modin_frame
    ._partitions
    .flatten()
)

partition_refs = [
    partition._data
    for partition in partitions
    if hasattr(partition, "_data")
]

results = ray.get([
    process_partition.remote(ref)
    for ref in partition_refs
])

The worker receives the underlying Pandas partition:

Modin DataFrame
      |
      +-- partition ObjectRef
      +-- partition ObjectRef
      +-- ...
      |
      v
Ray tasks
      |
      v
pandas.DataFrame partitions

In my test this was the fastest approach, at approximately 0.06 seconds, because there is no need to reconstruct a Modin DataFrame on each worker.

Again, the example accesses Modin private internals, so this should be considered a low-level workaround rather than a stable public API.

---

9. Option 3 — Use Ray Data for Ray-Native Pipelines

If the larger application is primarily a Ray pipeline with multiple Actors/Tasks, another option is to change the distributed data abstraction rather than trying to preserve a Modin DataFrame across Ray tasks.

Modin provides a Ray-specific conversion API for this:

ds = m_df.modin.to_ray()

For example:

@ray.remote
def process_dataset(dataset):
    return type(dataset), dataset.count()

ds = m_df.modin.to_ray()

dataset_type, count = ray.get(
    process_dataset.remote(ds)
)

print(dataset_type)
# <class 'ray.data.dataset.MaterializedDataset'>

print(count)
# 3

Here the worker receives a Ray Dataset rather than a Modin DataFrame.

The important point is that this does not solve "keep my Modin DataFrame". Instead, it uses Ray's own distributed data abstraction for a Ray-native pipeline.

Also, the benchmark above should not be interpreted as the intrinsic performance of Ray Data. In this reproduction, the conversion explicitly calls:

m_df._to_pandas()

followed by:

ray.data.from_pandas(...)

because m_df.modin.to_ray() is the Modin-provided conversion path for a Ray-backed Modin DataFrame.

Therefore, the ~5.33 second timing from my original benchmark should not be interpreted as the intrinsic performance of Ray Data. That particular benchmark included the explicit Pandas materialization/conversion step.

---

10. Recommendation

So, depending on the actual requirement:

• If the worker must receive a modin.pandas.DataFrame: the custom __reduce__ wrapper demonstrates a way to automate the manual reconstruction. It directly addresses the original question, but it depends on Modin internals.
• If the worker only needs to process the underlying data in parallel: passing the partition ObjectRefs is much faster in this reproduction.
• If the application is fundamentally a Ray-native distributed data pipeline: consider using ray.data.Dataset instead of trying to preserve the Modin DataFrame abstraction across Ray tasks.
• If long-term compatibility is important: I would avoid depending on modin.core.* and private attributes such as _query_compiler and _partitions unless Modin provides a corresponding public API.

The key point is that wrapping the DataFrame in another ordinary Python class does not preserve the Modin type. To keep a Modin DataFrame, the object has to be explicitly reconstructed on the worker; alternatively, the application can operate on the underlying partitions or use a Ray-native dataset abstraction.

I hope this helps clarify the serialization behavior and the available workarounds. If this doesn't fully resolve the issue in your setup, feel free to let me know and I'll be happy to look into it further. If it does resolve the issue, feel free to close the issue.