Live Notebook

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.

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.7/site-packages/IPython/core/display.py:694: UserWarning: Consider using IPython.display.IFrame instead
  warnings.warn("Consider using IPython.display.IFrame instead")

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
client = Client(n_workers=1, threads_per_worker=4, processes=False, memory_limit='2GB')
client

Client Scheduler: inproc://10.20.1.243/12158/1 Dashboard: http://10.20.1.243/12158/1: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.

import dask
df = dask.datasets.timeseries()
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

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.

!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

!head data/2000-01-01.csv

timestamp,id,name,x,y
2000-01-01 00:00:00,973,George,0.5323545272352643,0.4644479832473165
2000-01-01 00:00:01,950,Dan,-0.9742954488167075,0.39414517346882416
2000-01-01 00:00:02,1048,Zelda,0.5899397997531153,0.6707322213983693
2000-01-01 00:00:03,1025,Norbert,0.5806900810916886,-0.5177309147784845
2000-01-01 00:00:04,996,Frank,0.2640716830009191,-0.4828684051716825
2000-01-01 00:00:05,984,Frank,-0.4961715300024476,0.5894806879815138
2000-01-01 00:00:06,997,Quinn,0.10571372150947567,-0.8994397402222947
2000-01-01 00:00:07,1027,Xavier,-0.9209417344996893,0.030491890414622524
2000-01-01 00:00:08,956,Patricia,0.9196800400354601,0.6237577245387584

!head data/2000-01-30.csv

timestamp,id,name,x,y
2000-01-30 00:00:00,1017,Edith,0.15268171629288552,-0.8928227494533174
2000-01-30 00:00:01,973,Kevin,0.08001275540323682,-0.27745657835801807
2000-01-30 00:00:02,1024,Zelda,0.5770946221964139,0.5992811330713765
2000-01-30 00:00:03,952,Hannah,-0.4132934412479019,0.5760364741874113
2000-01-30 00:00:04,1009,Edith,-0.45565804239469565,-0.33539646668962875
2000-01-30 00:00:05,1046,Dan,0.43906296109467857,-0.5733613859942466
2000-01-30 00:00:06,976,Dan,-0.7296359315863512,-0.6606432280271826
2000-01-30 00:00:07,948,Oliver,-0.9769215439215688,0.27323227185668997
2000-01-30 00:00:08,1043,Sarah,0.3458069340849539,-0.7047037148329118

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

import pandas as pd

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

	timestamp	id	name	x	y
0	2000-01-01 00:00:00	973	George	0.532355	0.464448
1	2000-01-01 00:00:01	950	Dan	-0.974295	0.394145
2	2000-01-01 00:00:02	1048	Zelda	0.589940	0.670732
3	2000-01-01 00:00:03	1025	Norbert	0.580690	-0.517731
4	2000-01-01 00:00:04	996	Frank	0.264072	-0.482868

import dask.dataframe as dd

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

Dask DataFrame Structure:

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

Dask Name: from-delayed, 90 tasks

df.head()

	timestamp	id	name	x	y
0	2000-01-01 00:00:00	973	George	0.532355	0.464448
1	2000-01-01 00:00:01	950	Dan	-0.974295	0.394145
2	2000-01-01 00:00:02	1048	Zelda	0.589940	0.670732
3	2000-01-01 00:00:03	1025	Norbert	0.580690	-0.517731
4	2000-01-01 00:00:04	996	Frank	0.264072	-0.482868

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.

pd.read_csv?

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].

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

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.

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

CPU times: user 5.82 s, sys: 659 ms, total: 6.48 s
Wall time: 4.42 s

name
Alice      -0.002870
Bob         0.000055
Charlie    -0.000037
Dan        -0.001819
Edith       0.001411
Frank      -0.002131
George     -0.000985
Hannah     -0.000370
Ingrid     -0.000146
Jerry       0.001894
Kevin      -0.001100
Laura      -0.001183
Michael     0.002263
Norbert    -0.000146
Oliver     -0.000851
Patricia   -0.001639
Quinn      -0.001100
Ray         0.000625
Sarah      -0.001960
Tim         0.000302
Ursula     -0.002604
Victor      0.001306
Wendy      -0.000243
Xavier     -0.001541
Yvonne      0.000035
Zelda      -0.001988
Name: x, dtype: float64

Write to Parquet

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

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

!ls data/2000-01.parquet/

_common_metadata  part.15.parquet  part.22.parquet  part.2.parquet
_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

Read from Parquet

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

Dask DataFrame Structure:

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

Dask Name: read-parquet, 30 tasks

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

CPU times: user 1.79 s, sys: 121 ms, total: 1.91 s
Wall time: 1.36 s

name
Alice      -0.002870
Bob         0.000055
Charlie    -0.000037
Dan        -0.001819
Edith       0.001411
Frank      -0.002131
George     -0.000985
Hannah     -0.000370
Ingrid     -0.000146
Jerry       0.001894
Kevin      -0.001100
Laura      -0.001183
Michael     0.002263
Norbert    -0.000146
Oliver     -0.000851
Patricia   -0.001639
Quinn      -0.001100
Ray         0.000625
Sarah      -0.001960
Tim         0.000302
Ursula     -0.002604
Victor      0.001306
Wendy      -0.000243
Xavier     -0.001541
Yvonne      0.000035
Zelda      -0.001988
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.

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

CPU times: user 1.55 s, sys: 88.1 ms, total: 1.63 s
Wall time: 1.23 s

name
Alice      -0.002870
Bob         0.000055
Charlie    -0.000037
Dan        -0.001819
Edith       0.001411
Frank      -0.002131
George     -0.000985
Hannah     -0.000370
Ingrid     -0.000146
Jerry       0.001894
Kevin      -0.001100
Laura      -0.001183
Michael     0.002263
Norbert    -0.000146
Oliver     -0.000851
Patricia   -0.001639
Quinn      -0.001100
Ray         0.000625
Sarah      -0.001960
Tim         0.000302
Ursula     -0.002604
Victor      0.001306
Wendy      -0.000243
Xavier     -0.001541
Yvonne      0.000035
Zelda      -0.001988
Name: x, dtype: float64

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