You can run this notebook in a live session Binder 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&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")
[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

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

[3]:
import dask
df = dask.datasets.timeseries()
df
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/dask/dataframe/io/demo.py:91: FutureWarning: Creating a DatetimeIndex by passing range endpoints is deprecated.  Use `pandas.date_range` instead.
  freq=partition_freq))
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/dask/dataframe/io/demo.py:45: FutureWarning: Creating a DatetimeIndex by passing range endpoints is deprecated.  Use `pandas.date_range` instead.
  index = pd.DatetimeIndex(start=start, end=end, freq=freq, name='timestamp')
[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
[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,995,Charlie,-0.04830103453113721,-0.06269239255401637
2000-01-01 00:00:01,1000,Kevin,0.6451744312860419,0.26555308752977
2000-01-01 00:00:02,988,Wendy,0.12657599799215546,0.5210346023840606
2000-01-01 00:00:03,928,Edith,0.32105434295610813,0.43824591744794095
2000-01-01 00:00:04,1035,Tim,0.446865940927182,0.8583217649116621
2000-01-01 00:00:05,1006,Xavier,0.6103761238471914,-0.2959969231848809
2000-01-01 00:00:06,948,Xavier,0.7821293967826797,0.8267848903627324
2000-01-01 00:00:07,981,Quinn,-0.8318920549105977,0.2035839358219429
2000-01-01 00:00:08,973,Dan,-0.961633070243533,0.572344652562228
[7]:
!head data/2000-01-30.csv
timestamp,id,name,x,y
2000-01-30 00:00:00,999,Victor,-0.37850453433456344,0.9075572904272449
2000-01-30 00:00:01,960,Frank,0.515587177697896,0.8575461754247256
2000-01-30 00:00:02,1007,Bob,-0.21283513892050454,-0.3752556442143862
2000-01-30 00:00:03,990,Yvonne,-0.468159719710888,-0.4722855505825774
2000-01-30 00:00:04,996,Tim,0.9002137269281385,0.8447880434886135
2000-01-30 00:00:05,976,Xavier,-0.06239628080642179,0.40949819613460003
2000-01-30 00:00:06,936,Quinn,0.34534684643713964,0.9441933907973628
2000-01-30 00:00:07,1009,Oliver,-0.5983360028360669,-0.6635801459773993
2000-01-30 00:00:08,1039,Zelda,-0.4672665100129718,-0.9904618532585379

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 995 Charlie -0.048301 -0.062692
1 2000-01-01 00:00:01 1000 Kevin 0.645174 0.265553
2 2000-01-01 00:00:02 988 Wendy 0.126576 0.521035
3 2000-01-01 00:00:03 928 Edith 0.321054 0.438246
4 2000-01-01 00:00:04 1035 Tim 0.446866 0.858322
[9]:
import dask.dataframe as dd

df = dd.read_csv('data/2000-*-*.csv')
df
[9]:
Dask DataFrame Structure:
timestamp id name x y
npartitions=30
object int64 object float64 float64
... ... ... ... ...
... ... ... ... ... ...
... ... ... ... ...
... ... ... ... ...
Dask Name: from-delayed, 90 tasks
[10]:
df.head()
[10]:
timestamp id name x y
0 2000-01-01 00:00:00 995 Charlie -0.048301 -0.062692
1 2000-01-01 00:00:01 1000 Kevin 0.645174 0.265553
2 2000-01-01 00:00:02 988 Wendy 0.126576 0.521035
3 2000-01-01 00:00:03 928 Edith 0.321054 0.438246
4 2000-01-01 00:00:04 1035 Tim 0.446866 0.858322

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

[14]:
%time df.groupby('name').x.mean().compute()
CPU times: user 7.29 s, sys: 596 ms, total: 7.88 s
Wall time: 5.49 s
[14]:
name
Alice       0.001131
Bob        -0.002138
Charlie     0.002274
Dan        -0.001809
Edith      -0.001062
Frank      -0.001257
George      0.000149
Hannah     -0.001806
Ingrid     -0.002971
Jerry       0.003699
Kevin       0.001905
Laura       0.002698
Michael     0.000620
Norbert    -0.003655
Oliver      0.000540
Patricia   -0.000052
Quinn       0.004228
Ray        -0.001190
Sarah      -0.000231
Tim        -0.000930
Ursula     -0.002886
Victor     -0.000042
Wendy       0.000191
Xavier      0.001522
Yvonne      0.001249
Zelda      -0.000799
Name: x, dtype: float64
[15]:

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

[17]:
df = dd.read_parquet('data/2000-01.parquet', engine='pyarrow')
df
[17]:
Dask DataFrame Structure:
timestamp id name x y
npartitions=30
datetime64[ns] int64 object float64 float64
... ... ... ... ...
... ... ... ... ... ...
... ... ... ... ...
... ... ... ... ...
Dask Name: read-parquet, 30 tasks
[18]:
%time df.groupby('name').x.mean().compute()
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:707: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels = getattr(columns, 'labels', None) or [
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:734: FutureWarning: the 'labels' keyword is deprecated, use 'codes' instead
  return pd.MultiIndex(levels=new_levels, labels=labels, names=columns.names)
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:751: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels, = index.labels
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:707: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels = getattr(columns, 'labels', None) or [
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:734: FutureWarning: the 'labels' keyword is deprecated, use 'codes' instead
  return pd.MultiIndex(levels=new_levels, labels=labels, names=columns.names)
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:751: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels, = index.labels
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:707: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels = getattr(columns, 'labels', None) or [
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:734: FutureWarning: the 'labels' keyword is deprecated, use 'codes' instead
  return pd.MultiIndex(levels=new_levels, labels=labels, names=columns.names)
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:751: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels, = index.labels
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:707: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels = getattr(columns, 'labels', None) or [
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:734: FutureWarning: the 'labels' keyword is deprecated, use 'codes' instead
  return pd.MultiIndex(levels=new_levels, labels=labels, names=columns.names)
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:751: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels, = index.labels
CPU times: user 1.84 s, sys: 328 ms, total: 2.17 s
Wall time: 1.54 s
[18]:
name
Alice       0.001131
Bob        -0.002138
Charlie     0.002274
Dan        -0.001809
Edith      -0.001062
Frank      -0.001257
George      0.000149
Hannah     -0.001806
Ingrid     -0.002971
Jerry       0.003699
Kevin       0.001905
Laura       0.002698
Michael     0.000620
Norbert    -0.003655
Oliver      0.000540
Patricia   -0.000052
Quinn       0.004228
Ray        -0.001190
Sarah      -0.000231
Tim        -0.000930
Ursula     -0.002886
Victor     -0.000042
Wendy       0.000191
Xavier      0.001522
Yvonne      0.001249
Zelda      -0.000799
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()
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:707: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels = getattr(columns, 'labels', None) or [
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:734: FutureWarning: the 'labels' keyword is deprecated, use 'codes' instead
  return pd.MultiIndex(levels=new_levels, labels=labels, names=columns.names)
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:751: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels, = index.labels
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:707: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels = getattr(columns, 'labels', None) or [
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:734: FutureWarning: the 'labels' keyword is deprecated, use 'codes' instead
  return pd.MultiIndex(levels=new_levels, labels=labels, names=columns.names)
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:751: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels, = index.labels
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:707: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels = getattr(columns, 'labels', None) or [
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:734: FutureWarning: the 'labels' keyword is deprecated, use 'codes' instead
  return pd.MultiIndex(levels=new_levels, labels=labels, names=columns.names)
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:751: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels, = index.labels
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:707: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels = getattr(columns, 'labels', None) or [
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:734: FutureWarning: the 'labels' keyword is deprecated, use 'codes' instead
  return pd.MultiIndex(levels=new_levels, labels=labels, names=columns.names)
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:751: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels, = index.labels
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:707: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels = getattr(columns, 'labels', None) or [
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:734: FutureWarning: the 'labels' keyword is deprecated, use 'codes' instead
  return pd.MultiIndex(levels=new_levels, labels=labels, names=columns.names)
/home/travis/miniconda/envs/test/lib/python3.6/site-packages/pyarrow/pandas_compat.py:751: FutureWarning: .labels was deprecated in version 0.24.0. Use .codes instead.
  labels, = index.labels
CPU times: user 1.68 s, sys: 212 ms, total: 1.9 s
Wall time: 1.43 s

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