FEAT-#6970: Implement to/from_ray_dataset functions

docs/ecosystem.rst

@@ -45,5 +45,19 @@ where NumPy can be used and what libraries it powers.
 
     numpy_arr = to_numpy(modin_df)
 
+to_ray_dataset
+--------------
+
+You can refer to `Ray Data`_ page to get more details on
+where Ray Dataset can be used and what libraries it powers.
+
+.. code-block:: python
+
+    from modin.pandas.io import to_ray_dataset
+
+    ray_dataset = to_ray_dataset(modin_df)
+
 .. _pandas ecosystem: https://pandas.pydata.org/community/ecosystem.html
 .. _NumPy ecosystem: https://numpy.org
+.. _Ray Data: https://docs.ray.io/en/latest/data/data.html
+

modin/core/execution/dispatching/factories/dispatcher.py

@@ -181,6 +181,11 @@
     def from_dataframe(cls, *args, **kwargs):
         return cls.get_factory()._from_dataframe(*args, **kwargs)
 
+    @classmethod
+    @_inherit_docstrings(factories.BaseFactory._from_ray_dataset)
+    def from_ray_dataset(cls, ray_obj):
+        return cls.get_factory()._from_ray_dataset(ray_obj)
+
     @classmethod
     @_inherit_docstrings(factories.BaseFactory._read_parquet)
     def read_parquet(cls, **kwargs):
@@ -350,3 +355,8 @@
     @_inherit_docstrings(factories.BaseFactory._to_parquet)
     def to_parquet(cls, *args, **kwargs):
         return cls.get_factory()._to_parquet(*args, **kwargs)
+
+    @classmethod
+    @_inherit_docstrings(factories.BaseFactory._to_ray_dataset)
+    def to_ray_dataset(cls, modin_obj):
+        return cls.get_factory()._to_ray_dataset(modin_obj)

modin/core/execution/dispatching/factories/factories.py

@@ -201,6 +201,16 @@
     def _from_dataframe(cls, *args, **kwargs):
         return cls.io_cls.from_dataframe(*args, **kwargs)
 
+    @classmethod
+    @doc(
+        _doc_io_method_template,
+        source="a Ray Dataset",
+        params="ray_obj : ray.data.Dataset",
+        method="modin.core.execution.ray.implementations.pandas_on_ray.io.PandasOnRayIO.from_ray_dataset",
+    )
+    def _from_ray_dataset(cls, ray_obj):
+        return cls.io_cls.from_ray_dataset(ray_obj)
+
     @classmethod
     @doc(
         _doc_io_method_template,
@@ -455,6 +465,27 @@
         """
         return cls.io_cls.to_parquet(*args, **kwargs)
 
+    @classmethod
+    def _to_ray_dataset(cls, modin_obj):
+        """
+        Write query compiler content to a Ray Dataset.
+
+        Parameters
+        ----------
+        modin_obj : modin.pandas.DataFrame, modin.pandas.Series
+            The Modin DataFrame/Series to write.
+
+        Returns
+        -------
+        ray.data.Dataset
+            A Ray Dataset object.
+
+        Notes
+        -----
+        Modin DataFrame/Series can only be converted to a Ray Dataset if Modin uses a Ray engine.
+        """
+        return cls.io_cls.to_ray_dataset(modin_obj)
+
     # experimental methods that don't exist in pandas
     @classmethod
     @doc(

modin/core/execution/ray/generic/io/io.py

@@ -18,3 +18,51 @@
 
 class RayIO(BaseIO):
     """Base class for doing I/O operations over Ray."""
+
+    @classmethod
+    def from_ray_dataset(cls, ray_obj):
+        """
+        Create a Modin `query_compiler` from a Ray Dataset.
+
+        Parameters
+        ----------
+        ray_obj : ray.data.Dataset
+            The Ray Dataset to convert from.
+
+        Returns
+        -------
+        BaseQueryCompiler
+            QueryCompiler containing data from the Ray Dataset.
+
+        Notes
+        -----
+        This function must be implemented in every subclass
+        otherwise NotImplementedError will be raised.
+        """
+        raise NotImplementedError(
+            f"Modin dataset can't be created from `ray.data.Dataset` using {cls}."
+        )
+
+    @classmethod
+    def to_ray_dataset(cls, modin_obj):
+        """
+        Convert a Modin DataFrame/Series to a Ray Dataset.
+
+        Parameters
+        ----------
+        modin_obj : modin.DataFrame, modin.Series
+            The Modin DataFrame/Series to convert.
+
+        Returns
+        -------
+        ray.data.Dataset
+            Converted object with type depending on input.
+
+        Notes
+        -----
+        This function must be implemented in every subclass
+        otherwise NotImplementedError will be raised.
+        """
+        raise NotImplementedError(
+            f"`ray.data.Dataset` can't be created from Modin DataFrame/Series using {cls}."
+        )

modin/core/execution/ray/implementations/pandas_on_ray/io/io.py

@@ -17,6 +17,7 @@
 
 import pandas
 from pandas.io.common import get_handle, stringify_path
+from ray.data import from_pandas_refs
 
 from modin.core.execution.ray.common import RayWrapper, SignalActor
 from modin.core.execution.ray.generic.io import RayIO
@@ -258,3 +259,43 @@
         RayWrapper.materialize(
             [part.list_of_blocks[0] for row in result for part in row]
         )
+
+    @classmethod
+    def from_ray_dataset(cls, ray_obj):
+        """
+        Create a Modin `query_compiler` from a Ray Dataset.
+
+        Parameters
+        ----------
+        ray_obj : ray.data.Dataset
+            The Ray Dataset to convert from.
+
+        Returns
+        -------
+        BaseQueryCompiler
+            QueryCompiler containing data from the Ray Dataset.
+        """
+        from modin.distributed.dataframe.pandas.partitions import from_partitions
+
+        pd_objs = ray_obj.to_pandas_refs()
+        return from_partitions(pd_objs, axis=0)._query_compiler
+
+    @classmethod
+    def to_ray_dataset(cls, modin_obj):
+        """
+        Convert a Modin DataFrame/Series to a Ray Dataset.
+
+        Parameters
+        ----------
+        modin_obj : modin.DataFrame, modin.Series
+            The Modin DataFrame/Series to convert.
+
+        Returns
+        -------
+        ray.data.Dataset
+            Converted object with type depending on input.
+        """
+        from modin.distributed.dataframe.pandas.partitions import unwrap_partitions
+
+        parts = unwrap_partitions(modin_obj, axis=0)
+        return from_pandas_refs(parts)

modin/core/io/io.py

@@ -116,6 +116,30 @@
         """
         return cls.query_compiler_cls.from_dataframe(df, cls.frame_cls)
 
+    @classmethod
+    def from_ray_dataset(cls, ray_obj):
+        """
+        Create a Modin `query_compiler` from a Ray Dataset.
+
+        Parameters
+        ----------
+        ray_obj : ray.data.Dataset
+            The Ray Dataset to convert from.
+
+        Returns
+        -------
+        BaseQueryCompiler
+            QueryCompiler containing data from the Ray Dataset.
+
+        Notes
+        -----
+        Ray Dataset can only be converted to a Modin Dataframe if Modin uses a Ray engine.
+        If another engine is used, the runtime exception will be raised.
+        """
+        raise RuntimeError(
+            "Modin Dataframe can only be converted to a Ray Dataset if Modin uses a Ray engine."
+        )
+
     @classmethod
     @_inherit_docstrings(pandas.read_parquet, apilink="pandas.read_parquet")
     @doc(
@@ -694,3 +718,27 @@
             obj = obj.to_pandas()
 
         return obj.to_parquet(path, **kwargs)
+
+    @classmethod
+    def to_ray_dataset(cls, modin_obj):
+        """
+        Convert a Modin DataFrame/Series to a Ray Dataset.
+
+        Parameters
+        ----------
+        modin_obj : modin.DataFrame, modin.Series
+            The Modin DataFrame/Series to convert.
+
+        Returns
+        -------
+        ray.data.Dataset
+            Converted object with type depending on input.
+
+        Notes
+        -----
+        Modin DataFrame/Series can only be converted to a Ray Dataset if Modin uses a Ray engine.
+        If another engine is used, the runtime exception will be raised.
+        """
+        raise RuntimeError(
+            "Modin Dataframe can only be converted to a Ray Dataset if Modin uses a Ray engine."
+        )

modin/pandas/dataframe.py

@@ -44,7 +44,7 @@
 from modin.error_message import ErrorMessage
 from modin.logging import disable_logging
 from modin.pandas import Categorical
-from modin.pandas.io import from_non_pandas, from_pandas, to_pandas
+from modin.pandas.io import from_non_pandas, from_pandas, to_pandas, to_ray_dataset
 from modin.utils import (
     MODIN_UNNAMED_SERIES_LABEL,
     _inherit_docstrings,
@@ -2229,6 +2229,21 @@
             **kwargs,
         )
 
+    def to_ray_dataset(self):
+        """
+        Convert a Modin DataFrame to a Ray Dataset.
+
+        Returns
+        -------
+        ray.data.Dataset
+            Converted object with type depending on input.
+
+        Notes
+        -----
+        Modin Dataframe can only be converted to a Ray Dataset if Modin uses a Ray engine.
+        """
+        return to_ray_dataset(self)
+
     def to_period(
         self, freq=None, axis=0, copy=None
     ):  # pragma: no cover # noqa: PR01, RT01, D200

modin/pandas/io.py

@@ -1034,6 +1034,31 @@
     return ModinObjects.DataFrame(query_compiler=FactoryDispatcher.from_dataframe(df))
 
 
+def from_ray_dataset(ray_obj):
+    """
+    Convert a Ray Dataset into Modin DataFrame.
+
+    Parameters
+    ----------
+    ray_obj : ray.data.Dataset
+        The Ray Dataset to convert from.
+
+    Returns
+    -------
+    DataFrame
+        A new Modin DataFrame object.
+
+    Notes
+    -----
+    Ray Dataset can only be converted to Modin Dataframe if Modin uses a Ray engine.
+    """
+    from modin.core.execution.dispatching.factories.dispatcher import FactoryDispatcher
+
+    return ModinObjects.DataFrame(
+        query_compiler=FactoryDispatcher.from_ray_dataset(ray_obj)
+    )
+
+
 def to_pandas(modin_obj: SupportsPublicToPandas) -> Any:
     """
     Convert a Modin DataFrame/Series to a pandas DataFrame/Series.
@@ -1075,6 +1100,29 @@
     return array
 
 
+def to_ray_dataset(modin_obj):
+    """
+    Convert a Modin DataFrame/Series to a Ray Dataset.
+
+    Parameters
+    ----------
+    modin_obj : modin.DataFrame, modin.Series
+        The DataFrame/Series to convert.
+
+    Returns
+    -------
+    ray.data.Dataset
+        Converted object with type depending on input.
+
+    Notes
+    -----
+    Modin DataFrame/Series can only be converted to a Ray Dataset if Modin uses a Ray engine.
+    """
+    from modin.core.execution.dispatching.factories.dispatcher import FactoryDispatcher
+
+    return FactoryDispatcher.to_ray_dataset(modin_obj)
+
+
 __all__ = [
     "ExcelFile",
     "HDFStore",

modin/pandas/series.py

@@ -32,7 +32,7 @@
 
 from modin.config import PersistentPickle
 from modin.logging import disable_logging
-from modin.pandas.io import from_pandas, to_pandas
+from modin.pandas.io import from_pandas, to_pandas, to_ray_dataset
 from modin.utils import MODIN_UNNAMED_SERIES_LABEL, _inherit_docstrings
 
 from .accessor import CachedAccessor, SparseAccessor
@@ -1955,6 +1955,21 @@
 
     tolist = to_list
 
+    def to_ray_dataset(self):
+        """
+        Convert a Modin Series to a Ray Dataset.
+
+        Returns
+        -------
+        ray.data.Dataset
+            Converted object with type depending on input.
+
+        Notes
+        -----
+        Modin Series can only be converted to a Ray Dataset if Modin uses a Ray engine.
+        """
+        return to_ray_dataset(self)
+
     # TODO(williamma12): When we implement to_timestamp, have this call the version
     # in base.py
     def to_period(self, freq=None, copy=None):  # noqa: PR01, RT01, D200

modin/pandas/test/test_api.py

@@ -152,7 +152,7 @@ def test_dataframe_api_equality():
 
     ignore_in_pandas = ["timetuple"]
     # modin - namespace for using experimental functionality
-    ignore_in_modin = ["modin", "to_pandas"]
+    ignore_in_modin = ["modin", "to_pandas", "to_ray_dataset"]
     missing_from_modin = set(pandas_dir) - set(modin_dir)
     assert not len(
         missing_from_modin - set(ignore_in_pandas)
@@ -164,7 +164,7 @@ def test_dataframe_api_equality():
     ), "Differences found in API: {}".format(set(modin_dir) - set(pandas_dir))
 
     # These have to be checked manually
-    allowed_different = ["to_hdf", "hist", "modin", "to_pandas"]
+    allowed_different = ["to_hdf", "hist", "modin", "to_pandas", "to_ray_dataset"]
 
     assert_parameters_eq((pandas.DataFrame, pd.DataFrame), modin_dir, allowed_different)