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

DataFrames: Read and Write Data¶

Dask Dataframes can read and store data in many of the same formats as Pandas dataframes. In this example we read and write data with the popular CSV and Parquet formats, and discuss best practices when using these formats.

In [1]:

from IPython.display import HTML

HTML('<iframe width="560" height="315" src="https://www.youtube.com/embed/0eEsIA0O1iE?rel=0&amp;controls=0&amp;showinfo=0" frameborder="0" allowfullscreen></iframe>')

/home/travis/miniconda/envs/test/lib/python3.6/site-packages/IPython/core/display.py:689: UserWarning: Consider using IPython.display.IFrame instead
  warnings.warn("Consider using IPython.display.IFrame instead")

Out[1]:

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.

In [2]:

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

Out[2]:

Client

Scheduler: inproc://10.20.4.17/8065/1
Dashboard: http://localhost:8787/status

Cluster

Workers: 1
Cores: 4
Memory: 2.00 GB

Create artificial dataset¶

First we create an artificial dataset and write it to many CSV files.

You don’t need to understand this section, we’re just creating a dataset for the rest of the notebook.

In [3]:

import dask
df = dask.datasets.timeseries()
df

Out[3]:

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

In [4]:

import os
import datetime

if not os.path.exists('data'):
    os.mkdir('data')

def name(i):
    """ Provide date for filename given index

    Examples
    --------
    >>> name(0)
    '2000-01-01'
    >>> name(10)
    '2000-01-11'
    """
    return str(datetime.date(2000, 1, 1) + i * datetime.timedelta(days=1))

df.to_csv('data/*.csv', name_function=name);

Read CSV files¶

We now have many CSV files in our data directory, one for each day in the month of January 2000. Each CSV file holds timeseries data for that day. We can read all of them as one logical dataframe using the dd.read_csv function with a glob string.

In [5]:

!ls data/*.csv | head

data/2000-01-01.csv
data/2000-01-02.csv
data/2000-01-03.csv
data/2000-01-04.csv
data/2000-01-05.csv
data/2000-01-06.csv
data/2000-01-07.csv
data/2000-01-08.csv
data/2000-01-09.csv
data/2000-01-10.csv

In [6]:

!head data/2000-01-01.csv

timestamp,id,name,x,y
2000-01-01 00:00:00,966,Patricia,-0.8084939870066306,-0.7710739294530797
2000-01-01 00:00:01,991,Charlie,-0.7945776320546729,-0.964224365628507
2000-01-01 00:00:02,972,Jerry,-0.31433450401086116,0.5761011184842018
2000-01-01 00:00:03,995,Kevin,-0.6740486607314147,-0.8132252476668265
2000-01-01 00:00:04,980,Hannah,0.012616857066028997,-0.39662196393027704
2000-01-01 00:00:05,950,Zelda,-0.7994168561245698,-0.9736960965532464
2000-01-01 00:00:06,1004,Laura,-0.7361684053748219,-0.6870298568967275
2000-01-01 00:00:07,1006,Edith,0.1144035619295074,-0.8389967321195322
2000-01-01 00:00:08,964,Alice,0.7044046196285507,-0.6784299889508858

In [7]:

!head data/2000-01-30.csv

timestamp,id,name,x,y
2000-01-30 00:00:00,985,Ingrid,-0.6427537429917249,0.8590611614597978
2000-01-30 00:00:01,1017,Ingrid,0.27258421767284813,0.677370286499851
2000-01-30 00:00:02,1004,Kevin,-0.020798591147440204,-0.9305609199539306
2000-01-30 00:00:03,1011,Hannah,0.22767744076402208,-0.37900872516849926
2000-01-30 00:00:04,936,George,-0.8600604018135654,0.4071770735275315
2000-01-30 00:00:05,989,Xavier,-0.3182731127321907,0.8849712909185561
2000-01-30 00:00:06,1042,Kevin,0.8155731835833879,-0.5404474716618333
2000-01-30 00:00:07,974,Victor,-0.9176872159663572,-0.6834177265926906
2000-01-30 00:00:08,1050,Frank,0.6705057427848988,0.8777418091815345

We can read one file with pandas.read_csv or many files with dask.dataframe.read_csv

In [8]:

import pandas as pd

df = pd.read_csv('data/2000-01-01.csv')
df.head()

Out[8]:

	timestamp	id	name	x	y
0	2000-01-01 00:00:00	966	Patricia	-0.808494	-0.771074
1	2000-01-01 00:00:01	991	Charlie	-0.794578	-0.964224
2	2000-01-01 00:00:02	972	Jerry	-0.314335	0.576101
3	2000-01-01 00:00:03	995	Kevin	-0.674049	-0.813225
4	2000-01-01 00:00:04	980	Hannah	0.012617	-0.396622

In [9]:

import dask.dataframe as dd

df = dd.read_csv('data/2000-*-*.csv')
df

Out[9]:

Dask DataFrame Structure:

	timestamp	id	name	x	y
npartitions=30
	object	int64	object	float64	float64
	...	...	...	...	...
...	...	...	...	...	...
	...	...	...	...	...
	...	...	...	...	...

Dask Name: from-delayed, 90 tasks

In [10]:

df.head()

Out[10]:

	timestamp	id	name	x	y
0	2000-01-01 00:00:00	966	Patricia	-0.808494	-0.771074
1	2000-01-01 00:00:01	991	Charlie	-0.794578	-0.964224
2	2000-01-01 00:00:02	972	Jerry	-0.314335	0.576101
3	2000-01-01 00:00:03	995	Kevin	-0.674049	-0.813225
4	2000-01-01 00:00:04	980	Hannah	0.012617	-0.396622

Tuning read_csv¶

The Pandas read_csv function has many options to help you parse files. The Dask version uses the Pandas function internally, and so supports many of the same options. You can use the ? operator to see the full documentation string.

In [11]:

pd.read_csv?

In [12]:

dd.read_csv?

In this case we use the parse_dates keyword to parse the timestamp column to be a datetime. This will make things more efficient in the future. Notice that the dtype of the timestamp column has changed from object to datetime64[ns].

In [13]:

df = dd.read_csv('data/2000-*-*.csv', parse_dates=['timestamp'])
df

Out[13]:

Dask DataFrame Structure:

	timestamp	id	name	x	y
npartitions=30
	datetime64[ns]	int64	object	float64	float64
	...	...	...	...	...
...	...	...	...	...	...
	...	...	...	...	...
	...	...	...	...	...

Dask Name: from-delayed, 90 tasks

Do a simple computation¶

Whenever we operate on our dataframe we read through all of our CSV data so that we don’t fill up RAM. This is very efficient for memory use, but reading through all of the CSV files every time can be slow.

In [14]:

%time df.groupby('name').x.mean().compute()

CPU times: user 6.72 s, sys: 516 ms, total: 7.24 s
Wall time: 4.8 s

Out[14]:

name
Alice      -0.000824
Bob         0.002829
Charlie     0.002991
Dan        -0.000555
Edith      -0.002074
Frank      -0.001394
George     -0.000501
Hannah     -0.000727
Ingrid     -0.002006
Jerry      -0.001193
Kevin      -0.000590
Laura      -0.000761
Michael     0.001220
Norbert    -0.000153
Oliver      0.002355
Patricia   -0.001196
Quinn      -0.001985
Ray        -0.000461
Sarah       0.000012
Tim         0.001353
Ursula      0.003912
Victor     -0.001516
Wendy       0.003660
Xavier     -0.001425
Yvonne     -0.004744
Zelda      -0.005541
Name: x, dtype: float64

In [15]:

Write to Parquet¶

Instead, we’ll store our data in Parquet, a format that is more efficient for computers to read and write.

In [15]:

df.to_parquet('data/2000-01.parquet', engine='pyarrow')

In [16]:

!ls data/2000-01.parquet/

_common_metadata  part.16.parquet  part.23.parquet  part.3.parquet
part.0.parquet    part.17.parquet  part.24.parquet  part.4.parquet
part.10.parquet   part.18.parquet  part.25.parquet  part.5.parquet
part.11.parquet   part.19.parquet  part.26.parquet  part.6.parquet
part.12.parquet   part.1.parquet   part.27.parquet  part.7.parquet
part.13.parquet   part.20.parquet  part.28.parquet  part.8.parquet
part.14.parquet   part.21.parquet  part.29.parquet  part.9.parquet
part.15.parquet   part.22.parquet  part.2.parquet

Read from Parquet¶

In [17]:

df = dd.read_parquet('data/2000-01.parquet', engine='pyarrow')
df

Out[17]:

Dask DataFrame Structure:

	timestamp	id	name	x	y
npartitions=30
	datetime64[ns]	int64	object	float64	float64
	...	...	...	...	...
...	...	...	...	...	...
	...	...	...	...	...
	...	...	...	...	...

Dask Name: read-parquet, 30 tasks

In [18]:

%time df.groupby('name').x.mean().compute()

CPU times: user 1.47 s, sys: 208 ms, total: 1.68 s
Wall time: 1.15 s

Out[18]:

name
Alice      -0.000824
Bob         0.002829
Charlie     0.002991
Dan        -0.000555
Edith      -0.002074
Frank      -0.001394
George     -0.000501
Hannah     -0.000727
Ingrid     -0.002006
Jerry      -0.001193
Kevin      -0.000590
Laura      -0.000761
Michael     0.001220
Norbert    -0.000153
Oliver      0.002355
Patricia   -0.001196
Quinn      -0.001985
Ray        -0.000461
Sarah       0.000012
Tim         0.001353
Ursula      0.003912
Victor     -0.001516
Wendy       0.003660
Xavier     -0.001425
Yvonne     -0.004744
Zelda      -0.005541
Name: x, dtype: float64

Select only the columns that you plan to use¶

Parquet is a column-store, which means that it can efficiently pull out only a few columns from your dataset. This is good because it helps to avoid unnecessary data loading.

In [19]:

%%time
df = dd.read_parquet('data/2000-01.parquet', columns=['name', 'x'], engine='pyarrow')
df.groupby('name').x.mean().compute()

CPU times: user 1.27 s, sys: 188 ms, total: 1.46 s
Wall time: 1.03 s

Here the difference is not that large, but with larger datasets this can save a great deal of time.

Learn more¶

http://dask.pydata.org/en/latest/dataframe-create.html