You can run this notebook in a live session or view it on Github.

Dask DataFrames

Dask Dataframes coordinate many Pandas dataframes, partitioned along an index. They support a large subset of the Pandas API.

Start Dask Client for Dashboard

Starting the Dask Client is optional. It will provide a dashboard which is useful to gain insight on the computation.

The link to the dashboard will become visible when you create the client below. We recommend having it open on one side of your screen while using your notebook on the other side. This can take some effort to arrange your windows, but seeing them both at the same is very useful when learning.

from dask.distributed import Client, progress
client = Client(n_workers=2, threads_per_worker=2, memory_limit='1GB')
client

/home/travis/miniconda/envs/test/lib/python3.6/site-packages/dask/config.py:168: YAMLLoadWarning: calling yaml.load() without Loader=... is deprecated, as the default Loader is unsafe. Please read https://msg.pyyaml.org/load for full details.
  data = yaml.load(f.read()) or {}
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/distributed/config.py:20: YAMLLoadWarning: calling yaml.load() without Loader=... is deprecated, as the default Loader is unsafe. Please read https://msg.pyyaml.org/load for full details.
  defaults = yaml.load(f)

Client Scheduler: tcp://127.0.0.1:35064 Dashboard: http://127.0.0.1:8787/status

Cluster Workers: 2 Cores: 4 Memory: 2.00 GB

Create Random Dataframe

We create a random timeseries of data with the following attributes:

1. It stores a record for every 10 seconds of the year 2000
2. It splits that year by month, keeping every month as a separate Pandas dataframe
3. Along with a datetime index it has columns for names, ids, and numeric values

This is a small dataset of about 240 MB. Increase the number of days or reduce the frequency to practice with a larger dataset.

import dask
import dask.dataframe as dd
df = dask.datasets.timeseries()

Unlike Pandas, Dask DataFrames are lazy and so no data is printed here.

df

Dask DataFrame Structure:

	id	name	x	y
npartitions=30
2000-01-01	int64	object	float64	float64
2000-01-02	...	...	...	...
...	...	...	...	...
2000-01-30	...	...	...	...
2000-01-31	...	...	...	...

Dask Name: make-timeseries, 30 tasks

But the column names and dtypes are known.

df.dtypes

id        int64
name     object
x       float64
y       float64
dtype: object

Some operations will automatically display the data.

import pandas as pd
pd.options.display.precision = 2
pd.options.display.max_rows = 10

df.head(3)

	id	name	x	y
timestamp
2000-01-01 00:00:00	1000	Sarah	0.26	0.94
2000-01-01 00:00:01	1006	Wendy	-0.10	-0.15
2000-01-01 00:00:02	962	Victor	-0.58	-0.73

Use Standard Pandas Operations

Most common Pandas operations operate identically on Dask dataframes

df2 = df[df.y > 0]
df3 = df2.groupby('name').x.std()
df3

/home/travis/miniconda/envs/test/lib/python3.6/site-packages/dask/dataframe/utils.py:391: FutureWarning: Creating a DatetimeIndex by passing range endpoints is deprecated.  Use `pandas.date_range` instead.
  tz=idx.tz, name=idx.name)

Dask Series Structure:
npartitions=1
    float64
        ...
Name: x, dtype: float64
Dask Name: sqrt, 157 tasks

Call .compute() when you want your result as a Pandas dataframe.

If you started Client() above then you may want to watch the status page during computation.

computed_df = df3.compute()
type(computed_df)

pandas.core.series.Series

computed_df

name
Alice      0.58
Bob        0.58
Charlie    0.58
Dan        0.58
Edith      0.58
           ...
Victor     0.58
Wendy      0.58
Xavier     0.58
Yvonne     0.58
Zelda      0.58
Name: x, Length: 26, dtype: float64

Persist data in memory

If you have the available RAM for your dataset then you can persist data in memory.

This allows future computations to be much faster.

df = df.persist()

Time Series Operations

Because we have a datetime index time-series operations work efficiently

%matplotlib inline

df[['x', 'y']].resample('1h').mean().head()

/home/travis/miniconda/envs/test/lib/python3.6/site-packages/dask/dataframe/utils.py:391: FutureWarning: Creating a DatetimeIndex by passing range endpoints is deprecated.  Use `pandas.date_range` instead.
  tz=idx.tz, name=idx.name)

	x	y
timestamp
2000-01-01 00:00:00	-1.23e-03	6.35e-03
2000-01-01 01:00:00	-1.17e-02	3.95e-03
2000-01-01 02:00:00	5.94e-03	-1.16e-03
2000-01-01 03:00:00	-1.11e-02	-8.56e-03
2000-01-01 04:00:00	-9.79e-03	-5.69e-03

df[['x', 'y']].resample('24h').mean().compute().plot()

<matplotlib.axes._subplots.AxesSubplot at 0x7fa19573dd30>

df[['x', 'y']].rolling(window='24h').mean().head()

/home/travis/miniconda/envs/test/lib/python3.6/site-packages/dask/dataframe/utils.py:391: FutureWarning: Creating a DatetimeIndex by passing range endpoints is deprecated.  Use `pandas.date_range` instead.
  tz=idx.tz, name=idx.name)

	x	y
timestamp
2000-01-01 00:00:00	0.26	0.94
2000-01-01 00:00:01	0.08	0.39
2000-01-01 00:00:02	-0.14	0.02
2000-01-01 00:00:03	-0.32	-0.11
2000-01-01 00:00:04	-0.29	-0.04

Random access is cheap along the index, but must still be computed.

df.loc['2000-01-05']

Dask DataFrame Structure:

	id	name	x	y
npartitions=1
2000-01-05 00:00:00.000000000	int64	object	float64	float64
2000-01-05 23:59:59.999999999	...	...	...	...

Dask Name: loc, 31 tasks

%time df.loc['2000-01-05'].compute()

CPU times: user 36 ms, sys: 8 ms, total: 44 ms
Wall time: 66.2 ms

	id	name	x	y
timestamp
2000-01-05 00:00:00	986	Charlie	-0.74	0.35
2000-01-05 00:00:01	1017	George	-0.13	-0.91
2000-01-05 00:00:02	1046	Quinn	-0.39	-0.61
2000-01-05 00:00:03	1017	Norbert	0.01	0.24
2000-01-05 00:00:04	1007	Zelda	-0.56	-0.75
...	...	...	...	...
2000-01-05 23:59:55	1060	Oliver	0.04	-0.06
2000-01-05 23:59:56	935	Dan	0.75	-0.54
2000-01-05 23:59:57	1048	Michael	-0.06	-0.82
2000-01-05 23:59:58	1058	Jerry	0.99	0.17
2000-01-05 23:59:59	1015	Zelda	0.98	-0.18

86400 rows × 4 columns

Set Index

Data is sorted by the index column. This allows for faster access, joins, groupby-apply operations, etc.. However sorting data can be costly to do in parallel, so setting the index is both important to do, but only infrequently.

df = df.set_index('name')
df

Dask DataFrame Structure:

	id	x	y
npartitions=30
Alice	int64	float64	float64
Alice	...	...	...
...	...	...	...
Zelda	...	...	...
Zelda	...	...	...

Dask Name: sort_index, 1200 tasks

Because computing this dataset is expensive and we can fit it in our available RAM, we persist the dataset to memory.

df = df.persist()

Dask now knows where all data lives, indexed cleanly by name. As a result oerations like random access are cheap and efficient

%time df.loc['Alice'].compute()

CPU times: user 764 ms, sys: 108 ms, total: 872 ms
Wall time: 4.18 s

	id	x	y
name
Alice	990	-7.00e-01	-0.18
Alice	964	-3.00e-01	0.64
Alice	948	-4.69e-01	0.71
Alice	1006	1.29e-01	-0.52
Alice	996	1.63e-01	-0.69
...	...	...	...
Alice	990	6.20e-01	0.34
Alice	1084	-5.04e-01	0.75
Alice	957	-8.96e-02	0.03
Alice	938	-1.29e-01	0.83
Alice	988	-3.55e-04	-0.76

99565 rows × 3 columns

Groupby Apply with Scikit-Learn

Now that our data is sorted by name we can easily do operations like random access on name, or groupby-apply with custom functions.

Here we train a different Scikit-Learn linear regression model on each name.

from  sklearn.linear_model import LinearRegression

def train(partition):
    est = LinearRegression()
    est.fit(partition[['x']].values, partition.y.values)
    return est

df.groupby('name').apply(train, meta=object).compute()

name
Alice      LinearRegression(copy_X=True, fit_intercept=Tr...
Bob        LinearRegression(copy_X=True, fit_intercept=Tr...
Charlie    LinearRegression(copy_X=True, fit_intercept=Tr...
Dan        LinearRegression(copy_X=True, fit_intercept=Tr...
Edith      LinearRegression(copy_X=True, fit_intercept=Tr...
                                 ...
Victor     LinearRegression(copy_X=True, fit_intercept=Tr...
Wendy      LinearRegression(copy_X=True, fit_intercept=Tr...
Xavier     LinearRegression(copy_X=True, fit_intercept=Tr...
Yvonne     LinearRegression(copy_X=True, fit_intercept=Tr...
Zelda      LinearRegression(copy_X=True, fit_intercept=Tr...
Length: 26, dtype: object