Support conversion to/from cudf in dask.dataframe.core.to_backend

python/dask_cudf/dask_cudf/backends.py

@@ -30,7 +30,7 @@
     make_meta_obj,
 )
 from dask.sizeof import sizeof as sizeof_dispatch
-from dask.utils import is_arraylike
+from dask.utils import is_arraylike, Dispatch
 
 import cudf
 from cudf.api.types import is_string_dtype
@@ -447,10 +447,26 @@ def _default_backend(func, *args, **kwargs):
 
 
 try:
+    from dask.dataframe.backends import DataFrameBackendEntrypoint, PandasBackendEntrypoint
 
-    # Define "cudf" backend engine to be registered with Dask
-    from dask.dataframe.backends import DataFrameBackendEntrypoint
+    if hasattr(PandasBackendEntrypoint, "to_backend_dispatch"):
+        to_pandas_dispatch = PandasBackendEntrypoint.to_backend_dispatch()
+
+        @to_pandas_dispatch.register((cudf.DataFrame, cudf.Series, cudf.Index))
+        def to_pandas_dispatch_from_cudf(data, **kwargs):
+            return data.to_pandas(**kwargs)
+
+    to_cudf_dispatch = Dispatch("to_cudf_dispatch")
+
+    @to_cudf_dispatch.register((pd.DataFrame, pd.Series, pd.Index))
+    def to_cudf_dispatch_from_pandas(data, **kwargs):
+        return cudf.from_pandas(data, **kwargs)
 
+    @to_cudf_dispatch.register((cudf.DataFrame, cudf.Series, cudf.Index))
+    def to_cudf_dispatch_from_cudf(data, **kwargs):
+        return data
+
+    # Define "cudf" backend engine to be registered with Dask
     class CudfBackendEntrypoint(DataFrameBackendEntrypoint):
         """Backend-entrypoint class for Dask-DataFrame
 
@@ -470,6 +486,17 @@ class CudfBackendEntrypoint(DataFrameBackendEntrypoint):
         <class 'dask_cudf.core.DataFrame'>
         """
 
+        @classmethod
+        def to_backend_dispatch(cls):
+            return to_cudf_dispatch
+
+        @classmethod
+        def to_backend(cls, data: dd.core._Frame, **kwargs):
+            if isinstance(data._meta, (cudf.DataFrame, cudf.Series, cudf.Index)):
+                # Already a cudf-backed collection
+                return data
+            return data.map_partitions(cls.to_backend_dispatch(), **kwargs)
+
         @staticmethod
         def from_dict(
             data,

python/dask_cudf/dask_cudf/tests/test_core.py

@@ -6,6 +6,7 @@
 import numpy as np
 import pandas as pd
 import pytest
+from packaging import version
 
 import dask
 from dask import dataframe as dd
@@ -31,6 +32,28 @@ def test_from_dict_backend_dispatch():
     dd.assert_eq(expect, ddf)
 
 
+def test_to_backend():
+    # Test DataFrame.to_backend
+    # Requires Dask>=2023.1.1
+    if version.parse(dask.__version__) < version.parse("2023.1.1"):
+        pytest.skip("to_backend not supported in this version of Dask.")
+
+    np.random.seed(0)
+    data = {
+        "x": np.random.randint(0, 5, size=10000),
+        "y": np.random.normal(size=10000),
+    }
+    with dask.config.set({"dataframe.backend": "pandas"}):
+        ddf = dd.from_dict(data, npartitions=2)
+        assert isinstance(ddf._meta, pd.DataFrame)
+
+        gdf = ddf.to_backend("cudf")
+        assert isinstance(gdf, dgd.DataFrame)
+        dd.assert_eq(cudf.DataFrame(data), ddf)
+
+        assert isinstance(gdf.to_backend()._meta, pd.DataFrame)
+
+
 def test_from_cudf():
     np.random.seed(0)
 
@@ -547,8 +570,6 @@ def test_unary_ops(func, gdf, gddf):
 
     # Fixed in https://github.com/dask/dask/pull/4657
     if isinstance(p, cudf.Index):
-        from packaging import version
-
         if version.parse(dask.__version__) < version.parse("1.1.6"):
             pytest.skip(
                 "dask.dataframe assert_eq index check hardcoded to "