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.
[1]:
from IPython.display import HTML
HTML('<iframe width="560" height="315" src="https://www.youtube.com/embed/0eEsIA0O1iE?rel=0&controls=0&showinfo=0" frameborder="0" allowfullscreen></iframe>')
/usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages/IPython/core/display.py:717: UserWarning: Consider using IPython.display.IFrame instead
warnings.warn("Consider using IPython.display.IFrame instead")
[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.
[2]:
from dask.distributed import Client
client = Client(n_workers=1, threads_per_worker=4, processes=False, memory_limit='2GB')
client
[2]:
Client
|
Cluster
|
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.
[3]:
import dask
df = dask.datasets.timeseries()
df
[3]:
id | name | x | y | |
---|---|---|---|---|
npartitions=30 | ||||
2000-01-01 | int64 | object | float64 | float64 |
2000-01-02 | ... | ... | ... | ... |
... | ... | ... | ... | ... |
2000-01-30 | ... | ... | ... | ... |
2000-01-31 | ... | ... | ... | ... |
[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.
[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
[6]:
!head data/2000-01-01.csv
timestamp,id,name,x,y
2000-01-01 00:00:00,986,Yvonne,-0.46735166336132306,-0.5284651385903008
2000-01-01 00:00:01,1011,Dan,-0.6790377121970697,0.8700105617771643
2000-01-01 00:00:02,1013,Edith,0.418178614428417,-0.14669633816996352
2000-01-01 00:00:03,996,Victor,-0.5198557400468367,-0.8652698010069093
2000-01-01 00:00:04,996,Charlie,-0.9610545365394867,-0.9427385795794512
2000-01-01 00:00:05,990,Patricia,0.27569885239990777,-0.31787170745546645
2000-01-01 00:00:06,991,Dan,0.4957000956527089,-0.21546337885297828
2000-01-01 00:00:07,1015,Hannah,0.4916227762876131,0.8751971122211397
2000-01-01 00:00:08,983,Oliver,-0.8366077544280086,-0.21686128549035133
[7]:
!head data/2000-01-30.csv
timestamp,id,name,x,y
2000-01-30 00:00:00,987,Xavier,-0.9915377899142879,0.7908501574728211
2000-01-30 00:00:01,998,Patricia,-0.4263259190839772,0.11500038274372204
2000-01-30 00:00:02,965,Sarah,-0.6709731016475728,0.3869607902793153
2000-01-30 00:00:03,969,Wendy,-0.15867631498355084,0.4099494577986811
2000-01-30 00:00:04,964,Quinn,0.6233339948956649,-0.03215189901206794
2000-01-30 00:00:05,1058,Ingrid,-0.6432031748776097,0.7886701836849246
2000-01-30 00:00:06,1027,Victor,0.9508744730823087,0.8786747914899862
2000-01-30 00:00:07,975,Tim,0.6181908287513274,0.28870525524239965
2000-01-30 00:00:08,975,Edith,0.9633916421446942,0.7811130309205616
We can read one file with pandas.read_csv
or many files with dask.dataframe.read_csv
[8]:
import pandas as pd
df = pd.read_csv('data/2000-01-01.csv')
df.head()
[8]:
timestamp | id | name | x | y | |
---|---|---|---|---|---|
0 | 2000-01-01 00:00:00 | 986 | Yvonne | -0.467352 | -0.528465 |
1 | 2000-01-01 00:00:01 | 1011 | Dan | -0.679038 | 0.870011 |
2 | 2000-01-01 00:00:02 | 1013 | Edith | 0.418179 | -0.146696 |
3 | 2000-01-01 00:00:03 | 996 | Victor | -0.519856 | -0.865270 |
4 | 2000-01-01 00:00:04 | 996 | Charlie | -0.961055 | -0.942739 |
[9]:
import dask.dataframe as dd
df = dd.read_csv('data/2000-*-*.csv')
df
[9]:
timestamp | id | name | x | y | |
---|---|---|---|---|---|
npartitions=30 | |||||
object | int64 | object | float64 | float64 | |
... | ... | ... | ... | ... | |
... | ... | ... | ... | ... | ... |
... | ... | ... | ... | ... | |
... | ... | ... | ... | ... |
[10]:
df.head()
[10]:
timestamp | id | name | x | y | |
---|---|---|---|---|---|
0 | 2000-01-01 00:00:00 | 986 | Yvonne | -0.467352 | -0.528465 |
1 | 2000-01-01 00:00:01 | 1011 | Dan | -0.679038 | 0.870011 |
2 | 2000-01-01 00:00:02 | 1013 | Edith | 0.418179 | -0.146696 |
3 | 2000-01-01 00:00:03 | 996 | Victor | -0.519856 | -0.865270 |
4 | 2000-01-01 00:00:04 | 996 | Charlie | -0.961055 | -0.942739 |
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.
[11]:
pd.read_csv?
[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]
.
[13]:
df = dd.read_csv('data/2000-*-*.csv', parse_dates=['timestamp'])
df
[13]:
timestamp | id | name | x | y | |
---|---|---|---|---|---|
npartitions=30 | |||||
datetime64[ns] | int64 | object | float64 | float64 | |
... | ... | ... | ... | ... | |
... | ... | ... | ... | ... | ... |
... | ... | ... | ... | ... | |
... | ... | ... | ... | ... |
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.
[14]:
%time df.groupby('name').x.mean().compute()
CPU times: user 3.77 s, sys: 467 ms, total: 4.23 s
Wall time: 3.67 s
[14]:
name
Alice -0.001538
Bob -0.000560
Charlie -0.001016
Dan 0.001308
Edith 0.000495
Frank -0.002034
George -0.003631
Hannah 0.000827
Ingrid 0.000508
Jerry -0.006214
Kevin -0.001283
Laura 0.001582
Michael 0.000969
Norbert 0.002641
Oliver -0.000906
Patricia 0.000762
Quinn -0.001863
Ray -0.001239
Sarah -0.001611
Tim 0.000619
Ursula -0.001513
Victor -0.000515
Wendy -0.002153
Xavier -0.000080
Yvonne -0.000068
Zelda 0.001668
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.
[15]:
df.to_parquet('data/2000-01.parquet', engine='pyarrow')
[16]:
!ls data/2000-01.parquet/
_common_metadata part.14.parquet part.21.parquet part.29.parquet
_metadata part.15.parquet part.22.parquet part.3.parquet
part.0.parquet part.16.parquet part.23.parquet part.4.parquet
part.1.parquet part.17.parquet part.24.parquet part.5.parquet
part.10.parquet part.18.parquet part.25.parquet part.6.parquet
part.11.parquet part.19.parquet part.26.parquet part.7.parquet
part.12.parquet part.2.parquet part.27.parquet part.8.parquet
part.13.parquet part.20.parquet part.28.parquet part.9.parquet
Read from Parquet¶
[17]:
df = dd.read_parquet('data/2000-01.parquet', engine='pyarrow')
df
[17]:
timestamp | id | name | x | y | |
---|---|---|---|---|---|
npartitions=30 | |||||
datetime64[ns] | int64 | object | float64 | float64 | |
... | ... | ... | ... | ... | |
... | ... | ... | ... | ... | ... |
... | ... | ... | ... | ... | |
... | ... | ... | ... | ... |
[18]:
%time df.groupby('name').x.mean().compute()
CPU times: user 1.13 s, sys: 102 ms, total: 1.23 s
Wall time: 1.13 s
[18]:
name
Alice -0.001538
Bob -0.000560
Charlie -0.001016
Dan 0.001308
Edith 0.000495
Frank -0.002034
George -0.003631
Hannah 0.000827
Ingrid 0.000508
Jerry -0.006214
Kevin -0.001283
Laura 0.001582
Michael 0.000969
Norbert 0.002641
Oliver -0.000906
Patricia 0.000762
Quinn -0.001863
Ray -0.001239
Sarah -0.001611
Tim 0.000619
Ursula -0.001513
Victor -0.000515
Wendy -0.002153
Xavier -0.000080
Yvonne -0.000068
Zelda 0.001668
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.
[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.09 s, sys: 74.1 ms, total: 1.16 s
Wall time: 1.11 s
[19]:
name
Alice -0.001538
Bob -0.000560
Charlie -0.001016
Dan 0.001308
Edith 0.000495
Frank -0.002034
George -0.003631
Hannah 0.000827
Ingrid 0.000508
Jerry -0.006214
Kevin -0.001283
Laura 0.001582
Michael 0.000969
Norbert 0.002641
Oliver -0.000906
Patricia 0.000762
Quinn -0.001863
Ray -0.001239
Sarah -0.001611
Tim 0.000619
Ursula -0.001513
Victor -0.000515
Wendy -0.002153
Xavier -0.000080
Yvonne -0.000068
Zelda 0.001668
Name: x, dtype: float64
Here the difference is not that large, but with larger datasets this can save a great deal of time.