Simply use the `Dask configuration
<https://docs.dask.org/en/stable/how-to/selecting-the-collection-backend.html>`__
system to set the ``"dataframe.backend"`` option to ``"cudf"``.
From Python, this can be achieved like so::

  import dask

  dask.config.set({"dataframe.backend": "cudf"})

Alternatively, you can set ``DASK_DATAFRAME__BACKEND=cudf`` in the
environment before running your code.

Once this is done, the public Dask DataFrame API will leverage
``cudf`` automatically when a new DataFrame collection is created
from an on-disk format using any of the following ``dask.dataframe``
functions:

* :py:func:`dask.dataframe.read_parquet`
* :py:func:`dask.dataframe.read_json`
* :py:func:`dask.dataframe.read_csv`
* :py:func:`dask.dataframe.read_orc`
* :py:func:`dask.dataframe.read_hdf`
* :py:meth:`dask.dataframe.DataFrame.from_dict`


For example::

  import dask.dataframe as dd

  # By default, we obtain a pandas-backed dataframe
  df = dd.read_parquet("data.parquet", ...)

  import dask

  dask.config.set({"dataframe.backend": "cudf"})
  # This now gives us a cuDF-backed dataframe
  df = dd.read_parquet("data.parquet", ...)

When other functions are used to create a new collection
(e.g. :func:`dask.dataframe.from_map`, :func:`dask.dataframe.from_pandas`, :func:`dask.dataframe.from_delayed`,
and :func:`dask.dataframe.from_array`), the backend of the new collection will
depend on the inputs to those functions. For example::

  import pandas as pd
  import cudf

  # This gives us a pandas-backed dataframe
  dd.from_pandas(pd.DataFrame({"a": range(10)}))

  # This gives us a cuDF-backed dataframe
  dd.from_pandas(cudf.DataFrame({"a": range(10)}))

An existing collection can always be moved to a specific backend
using the :meth:`dask.dataframe.DataFrame.to_backend` API::

  # This ensures that we have a cuDF-backed dataframe
  df = df.to_backend("cudf")

  # This ensures that we have a pandas-backed dataframe
  df = df.to_backend("pandas")

The explicit Dask cuDF API
~~~~~~~~~~~~~~~~~~~~~~~~~~

In addition to providing the ``"cudf"`` backend for Dask DataFrame,
Dask cuDF also provides an explicit ``dask_cudf`` API::

  import dask_cudf

  # This always gives us a cuDF-backed dataframe
  df = dask_cudf.read_parquet("data.parquet", ...)

This API is used implicitly by the Dask DataFrame API when the ``"cudf"``
backend is enabled. Therefore, using it directly will not provide any
performance benefit over the CPU/GPU-portable ``dask.dataframe`` API.
Also, using some parts of the explicit API are incompatible with
automatic query planning (see the next section).

Query Planning
~~~~~~~~~~~~~~

Dask cuDF now provides automatic query planning by default (RAPIDS 24.06+).
As long as the ``"dataframe.query-planning"`` configuration is set to
``True`` (the default) when ``dask.dataframe`` is first imported, `Dask
Expressions <https://github.com/dask/dask-expr>`__ will be used under the hood.

For example, the following code will automatically benefit from predicate
pushdown when the result is computed::

  df = dd.read_parquet("/my/parquet/dataset/")
  result = df.sort_values('B')['A']

Unoptimized expression graph (``df.pprint()``)::

  Projection: columns='A'
    SortValues: by=['B'] shuffle_method='tasks' options={}
      ReadParquetFSSpec: path='/my/parquet/dataset/' ...

Simplified expression graph (``df.simplify().pprint()``)::

  Projection: columns='A'
    SortValues: by=['B'] shuffle_method='tasks' options={}
      ReadParquetFSSpec: path='/my/parquet/dataset/' columns=['A', 'B'] ...

.. note::
  Dask will automatically simplify the expression graph (within
  :func:`dask.optimize`) when the result is converted to a task graph
  (via :func:`dask.compute` or :func:`dask.persist`). You do not need
  to optimize or simplify the graph yourself.

Using Multiple GPUs and Multiple Nodes

client = Client(
  LocalCUDACluster(
    CUDA_VISIBLE_DEVICES="0,1",  # Use two workers (on devices 0 and 1)
    rmm_pool_size=0.9,  # Use 90% of GPU memory as a pool for faster allocations
    enable_cudf_spill=True,  # Improve device memory stability
    local_directory="/fast/scratch/",  # Use fast local storage for spilling
  )
)

df = dd.read_parquet("/my/parquet/dataset/")
agg = df.groupby('B').sum()
agg.compute()  # This will use the cluster defined above