Dask cuDF

https://docs.nvidia.com/dask-cudf/latest/
Admin
Dask cuDF is an extension library for the Dask parallel computing framework that provides...
Tokens:7,827
Snippets:71
Trust Score:10
Update:3 days ago
Show doc for...
Tokens:
Raw
### Construct a Dask DataFrame from a Pandas DataFrame

Source: https://docs.nvidia.com/dask-cudf/latest/api

Demonstrates creating a Dask DataFrame from a Pandas DataFrame or Series using from_pandas, including partitioning and index division examples.

```python
>>> from dask.dataframe import from_pandas
>>> df = pd.DataFrame(dict(a=list('aabbcc'), b=list(range(6))),
...                   index=pd.date_range(start='20100101', periods=6))
>>> ddf = from_pandas(df, npartitions=3)
>>> ddf.divisions
(Timestamp('2010-01-01 00:00:00'),
 Timestamp('2010-01-03 00:00:00'),
 Timestamp('2010-01-05 00:00:00'),
 Timestamp('2010-01-06 00:00:00'))
>>> ddf = from_pandas(df.a, npartitions=3)  # Works with Series too!
>>> ddf.divisions
(Timestamp('2010-01-01 00:00:00'),
 Timestamp('2010-01-03 00:00:00'),
 Timestamp('2010-01-05 00:00:00'),
 Timestamp('2010-01-06 00:00:00'))
```

--------------------------------

### Automatic query planning example

Source: https://docs.nvidia.com/dask-cudf/latest

Dask cuDF automatically applies optimizations like predicate pushdown when query planning is enabled.

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

--------------------------------

### Configure LocalCUDACluster and execute Dask workflow

Source: https://docs.nvidia.com/dask-cudf/latest

Initializes a Dask client with a LocalCUDACluster configured for two GPUs, memory pooling, and spill-to-disk capabilities. The workflow demonstrates reading a Parquet dataset, performing a groupby aggregation, and triggering computation.

```python
from dask_cuda import LocalCUDACluster
from distributed import Client

if __name__ == "__main__":

  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
```

--------------------------------

### View unoptimized expression graph

Source: https://docs.nvidia.com/dask-cudf/latest

Display the expression graph before optimization using the pprint method.

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

--------------------------------

### Convert collection backend using to_backend

Source: https://docs.nvidia.com/dask-cudf/latest

Use the to_backend method to explicitly move an existing collection to a specific backend.

```python
# 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")
```

--------------------------------

### View simplified expression graph

Source: https://docs.nvidia.com/dask-cudf/latest

Display the optimized expression graph after calling simplify and pprint.

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

--------------------------------

### Create DataFrames from existing collections

Source: https://docs.nvidia.com/dask-cudf/latest

The backend of a new collection created via from_pandas depends on the input type.

```python
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)}))
```

--------------------------------

### Create cuDF-backed DataFrames via Dask DataFrame API

Source: https://docs.nvidia.com/dask-cudf/latest

Demonstrates how setting the cudf backend changes the behavior of dask.dataframe functions like read_parquet.

```python
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", ...)
```

--------------------------------

### Configure Dask backend to use cudf

Source: https://docs.nvidia.com/dask-cudf/latest/api

Set the Dask dataframe backend to cudf to enable Dask-cuDF functionality.

```python
>>> import dask
>>> dask.config.set({"dataframe.backend": "cudf"})
```

--------------------------------

### Convert Dask DataFrame backend

Source: https://docs.nvidia.com/dask-cudf/latest/best_practices

Use the to_backend method to convert a Dask DataFrame to a different backend, such as pandas. Note that frequent CPU-GPU data movement can negatively impact performance.

```python
df = df.to_backend("pandas")  # This gives us a pandas-backed collection
```

--------------------------------

### Use the explicit dask_cudf API

Source: https://docs.nvidia.com/dask-cudf/latest

The dask_cudf module provides an explicit API that always returns a cuDF-backed dataframe.

```python
import dask_cudf

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

--------------------------------

### dask_cudf.from_cudf

Source: https://docs.nvidia.com/dask-cudf/latest/api

Creates a dask.dataframe.DataFrame from a cudf.DataFrame. This function is a thin wrapper around dask.dataframe.from_pandas() and accepts the same arguments, operating on cuDF objects.

```APIDOC
## dask_cudf.from_cudf

### Description
Creates a dask.dataframe.DataFrame from a cudf.DataFrame. This function is a thin wrapper around dask.dataframe.from_pandas() and accepts the same arguments, operating on cuDF objects.

### Method
Function call

### Endpoint
from_cudf(data, npartitions=None, chunksize=None, sort=True, name=None)

### Parameters
#### Positional Parameters
- **data** (pandas.DataFrame or pandas.Series) - Required - The DataFrame/Series with which to construct a Dask DataFrame/Series.

#### Keyword Parameters
- **npartitions** (int, optional, default 1) - Optional - The number of partitions of the index to create. Note that if there are duplicate values or insufficient elements in data.index, the output may have fewer partitions than requested.
- **chunksize** (int, optional) - Optional - The desired number of rows per index partition to use. Note that depending on the size and index of the dataframe, actual partition sizes may vary.
- **sort** (bool, default True) - Optional - Sort the input by index first to obtain cleanly divided partitions (with known divisions). If False, the input will not be sorted, and all divisions will be set to None.
- **name** (optional) - Optional - Name for the resulting Dask collection.

### Returns
- **dask.DataFrame or dask.Series** - A dask DataFrame/Series partitioned along the index.

### Raises
- **TypeError** - If something other than a pandas.DataFrame or pandas.Series is passed in.

### Example
```python
>>> from dask.dataframe import from_pandas
>>> df = pd.DataFrame(dict(a=list('aabbcc'), b=list(range(6))),
...                   index=pd.date_range(start='20100101', periods=6))
>>> ddf = from_pandas(df, npartitions=3)
>>> ddf.divisions
(Timestamp('2010-01-01 00:00:00'),
 Timestamp('2010-01-03 00:00:00'),
 Timestamp('2010-01-05 00:00:00'),
 Timestamp('2010-01-06 00:00:00'))
>>> ddf = from_pandas(df.a, npartitions=3)  # Works with Series too!
>>> ddf.divisions
(Timestamp('2010-01-01 00:00:00'),
 Timestamp('2010-01-03 00:00:00'),
 Timestamp('2010-01-05 00:00:00'),
 Timestamp('2010-01-06 00:00:00'))
```
```

--------------------------------

### Synchronize distributed persist with wait

Source: https://docs.nvidia.com/dask-cudf/latest/best_practices

Use the distributed.wait() function to create a blocking synchronization point after calling persist() on a Dask DataFrame.

```python
ddf = ddf.persist()
wait(ddf)
```
Tokens:
Raw