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

Dask DataFrames

We finished Chapter 1 by building a parallel dataframe computation over a directory of CSV files using dask.delayed. In this section we use dask.dataframe to automatically build similiar computations, for the common case of tabular computations. Dask dataframes look and feel like Pandas dataframes but they run on the same infrastructure that powers dask.delayed.

In this notebook we use the same airline data as before, but now rather than write for-loops we let dask.dataframe construct our computations for us. The dask.dataframe.read_csv function can take a globstring like "data/nycflights/*.csv" and build parallel computations on all of our data at once.

When to use dask.dataframe

Pandas is great for tabular datasets that fit in memory. Dask becomes useful when the dataset you want to analyze is larger than your machine’s RAM. The demo dataset we’re working with is only about 200MB, so that you can download it in a reasonable time, but dask.dataframe will scale to datasets much larger than memory.

The dask.dataframe module implements a blocked parallel DataFrame object that mimics a large subset of the Pandas DataFrame. One Dask DataFrame is comprised of many in-memory pandas DataFrames separated along the index. One operation on a Dask DataFrame triggers many pandas operations on the constituent pandas DataFrames in a way that is mindful of potential parallelism and memory constraints.

Related Documentation

• DataFrame documentation

• DataFrame screencast

• DataFrame API

• DataFrame examples

• Pandas documentation

Main Take-aways

• Dask DataFrame should be familiar to Pandas users

• The partitioning of dataframes is important for efficient execution

Create data

[1]:

%run prep.py -d flights

Setup

[2]:

from dask.distributed import Client
 
client = Client(n_workers=4)

We create artifical data.

[3]:

from prep import accounts_csvs
accounts_csvs()
 
import os
import dask
filename = os.path.join('data', 'accounts.*.csv')
filename

[3]:

'data/accounts.*.csv'

Filename includes a glob pattern *, so all files in the path matching that pattern will be read into the same Dask DataFrame.

[4]:

import dask.dataframe as dd
df = dd.read_csv(filename)
df.head()

[4]:

[5]:

# load and count number of rows
len(df)

[5]:

30000

What happened here? - Dask investigated the input path and found that there are three matching files - a set of jobs was intelligently created for each chunk - one per original CSV file in this case - each file was loaded into a pandas dataframe, had len() applied to it - the subtotals were combined to give you the final grand total.

Real Data

Lets try this with an extract of flights in the USA across several years. This data is specific to flights out of the three airports in the New York City area.

[6]:

df = dd.read_csv(os.path.join('data', 'nycflights', '*.csv'),
                 parse_dates={'Date': [0, 1, 2]})

Notice that the respresentation of the dataframe object contains no data - Dask has just done enough to read the start of the first file, and infer the column names and dtypes.

[7]:

df

[7]:

Dask DataFrame Structure:

Dask Name: from-delayed, 30 tasks

We can view the start and end of the data

[8]:

df.head()

[8]:

5 rows × 21 columns

[9]:

df.tail()  # this fails

---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
<ipython-input-9-430ef93b601c> in <module>
----> 1 df.tail()  # this fails
 
~/miniconda/envs/test/lib/python3.7/site-packages/dask/dataframe/core.py in tail(self, n, compute)
    996
    997         if compute:
--> 998             result = result.compute()
    999         return result
   1000
 
~/miniconda/envs/test/lib/python3.7/site-packages/dask/base.py in compute(self, **kwargs)
    163         dask.base.compute
    164         """
--> 165         (result,) = compute(self, traverse=False, **kwargs)
    166         return result
    167
 
~/miniconda/envs/test/lib/python3.7/site-packages/dask/base.py in compute(*args, **kwargs)
    434     keys = [x.__dask_keys__() for x in collections]
    435     postcomputes = [x.__dask_postcompute__() for x in collections]
--> 436     results = schedule(dsk, keys, **kwargs)
    437     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
    438
 
~/miniconda/envs/test/lib/python3.7/site-packages/distributed/client.py in get(self, dsk, keys, restrictions, loose_restrictions, resources, sync, asynchronous, direct, retries, priority, fifo_timeout, actors, **kwargs)
   2570                     should_rejoin = False
   2571             try:
-> 2572                 results = self.gather(packed, asynchronous=asynchronous, direct=direct)
   2573             finally:
   2574                 for f in futures.values():
 
~/miniconda/envs/test/lib/python3.7/site-packages/distributed/client.py in gather(self, futures, errors, direct, asynchronous)
   1870                 direct=direct,
   1871                 local_worker=local_worker,
-> 1872                 asynchronous=asynchronous,
   1873             )
   1874
 
~/miniconda/envs/test/lib/python3.7/site-packages/distributed/client.py in sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
    765         else:
    766             return sync(
--> 767                 self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
    768             )
    769
 
~/miniconda/envs/test/lib/python3.7/site-packages/distributed/utils.py in sync(loop, func, callback_timeout, *args, **kwargs)
    332     if error[0]:
    333         typ, exc, tb = error[0]
--> 334         raise exc.with_traceback(tb)
    335     else:
    336         return result[0]
 
~/miniconda/envs/test/lib/python3.7/site-packages/distributed/utils.py in f()
    316             if callback_timeout is not None:
    317                 future = gen.with_timeout(timedelta(seconds=callback_timeout), future)
--> 318             result[0] = yield future
    319         except Exception as exc:
    320             error[0] = sys.exc_info()
 
~/miniconda/envs/test/lib/python3.7/site-packages/tornado/gen.py in run(self)
    733
    734                     try:
--> 735                         value = future.result()
    736                     except Exception:
    737                         exc_info = sys.exc_info()
 
~/miniconda/envs/test/lib/python3.7/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
   1726                             exc = CancelledError(key)
   1727                         else:
-> 1728                             raise exception.with_traceback(traceback)
   1729                         raise exc
   1730                     if errors == "skip":
 
~/miniconda/envs/test/lib/python3.7/site-packages/dask/utils.py in apply()
     27 def apply(func, args, kwargs=None):
     28     if kwargs:
---> 29         return func(*args, **kwargs)
     30     else:
     31         return func(*args)
 
~/miniconda/envs/test/lib/python3.7/site-packages/dask/dataframe/io/csv.py in pandas_read_text()
     68     df = reader(bio, **kwargs)
     69     if dtypes:
---> 70         coerce_dtypes(df, dtypes)
     71
     72     if enforce and columns and (list(df.columns) != list(columns)):
 
~/miniconda/envs/test/lib/python3.7/site-packages/dask/dataframe/io/csv.py in coerce_dtypes()
    172             rule.join(filter(None, [dtype_msg, date_msg]))
    173         )
--> 174         raise ValueError(msg)
    175
    176
 
ValueError: Mismatched dtypes found in `pd.read_csv`/`pd.read_table`.
 
+----------------+---------+----------+
| Column         | Found   | Expected |
+----------------+---------+----------+
| CRSElapsedTime | float64 | int64    |
| TailNum        | object  | float64  |
+----------------+---------+----------+
 
The following columns also raised exceptions on conversion:
 
- TailNum
  ValueError("could not convert string to float: 'N54711'")
 
Usually this is due to dask's dtype inference failing, and
*may* be fixed by specifying dtypes manually by adding:
 
dtype={'CRSElapsedTime': 'float64',
       'TailNum': 'object'}
 
to the call to `read_csv`/`read_table`.

What just happened?

Unlike pandas.read_csv which reads in the entire file before inferring datatypes, dask.dataframe.read_csv only reads in a sample from the beginning of the file (or first file if using a glob). These inferred datatypes are then enforced when reading all partitions.

In this case, the datatypes inferred in the sample are incorrect. The first n rows have no value for CRSElapsedTime (which pandas infers as a float), and later on turn out to be strings (object dtype). Note that Dask gives an informative error message about the mismatch. When this happens you have a few options:

• Specify dtypes directly using the dtype keyword. This is the recommended solution, as it’s the least error prone (better to be explicit than implicit) and also the most performant.

• Increase the size of the sample keyword (in bytes)

• Use assume_missing to make dask assume that columns inferred to be int (which don’t allow missing values) are actually floats (which do allow missing values). In our particular case this doesn’t apply.

In our case we’ll use the first option and directly specify the dtypes of the offending columns.

[10]:

df = dd.read_csv(os.path.join('data', 'nycflights', '*.csv'),
                 parse_dates={'Date': [0, 1, 2]},
                 dtype={'TailNum': str,
                        'CRSElapsedTime': float,
                        'Cancelled': bool})

[11]:

df.tail()  # now works

[11]:

5 rows × 21 columns

Computations with dask.dataframe

We compute the maximum of the DepDelay column. With just pandas, we would loop over each file to find the individual maximums, then find the final maximum over all the individual maximums

maxes = []
for fn in filenames:
    df = pd.read_csv(fn)
    maxes.append(df.DepDelay.max())
 
final_max = max(maxes)

We could wrap that pd.read_csv with dask.delayed so that it runs in parallel. Regardless, we’re still having to think about loops, intermediate results (one per file) and the final reduction (max of the intermediate maxes). This is just noise around the real task, which pandas solves with

df = pd.read_csv(filename, dtype=dtype)
df.DepDelay.max()

dask.dataframe lets us write pandas-like code, that operates on larger than memory datasets in parallel.

[12]:

%time df.DepDelay.max().compute()

CPU times: user 73.1 ms, sys: 6.91 ms, total: 80 ms
Wall time: 702 ms

[12]:

409.0

This writes the delayed computation for us and then runs it.

Some things to note:

• As with dask.delayed, we need to call .compute() when we’re done. Up until this point everything is lazy.

• Dask will delete intermediate results (like the full pandas dataframe for each file) as soon as possible.

  • This lets us handle datasets that are larger than memory

  • This means that repeated computations will have to load all of the data in each time (run the code above again, is it faster or slower than you would expect?)

As with Delayed objects, you can view the underlying task graph using the .visualize method:

[13]:

# notice the parallelism
df.DepDelay.max().visualize()

[13]:

Exercises

In this section we do a few dask.dataframe computations. If you are comfortable with Pandas then these should be familiar. You will have to think about when to call compute.

1.) How many rows are in our dataset?

If you aren’t familiar with pandas, how would you check how many records are in a list of tuples?

[14]:

# Your code here

[15]:

len(df)

[15]:

9990

2.) In total, how many non-canceled flights were taken?

With pandas, you would use boolean indexing.

[16]:

# Your code here

[17]:

len(df[~df.Cancelled])

[17]:

9383

3.) In total, how many non-cancelled flights were taken from each airport?

Hint: use df.groupby</code> &lt;[https://pandas.pydata.org/pandas-docs/stable/groupby.html](https://pandas.pydata.org/pandas-docs/stable/groupby.html)&gt;__.

[18]:

# Your code here

[19]:

df[~df.Cancelled].groupby('Origin').Origin.count().compute()

[19]:

Origin
EWR    4132
JFK    1085
LGA    4166
Name: Origin, dtype: int64

[20]:

# Your code here

[21]:

df.groupby("Origin").DepDelay.mean().compute()

[21]:

Origin
EWR    12.500968
JFK    17.053456
LGA    10.169227
Name: DepDelay, dtype: float64

[22]:

# Your code here

[23]:

df.groupby("DayOfWeek").DepDelay.mean().compute()

[23]:

DayOfWeek
1    10.677698
2     8.633310
3    14.208160
4    14.187853
5    15.209929
6     9.540307
7    10.609375
Name: DepDelay, dtype: float64

[24]:

non_cancelled = df[~df.Cancelled]
mean_delay = non_cancelled.DepDelay.mean()
std_delay = non_cancelled.DepDelay.std()

[25]:

%%time
 
mean_delay_res = mean_delay.compute()
std_delay_res = std_delay.compute()

CPU times: user 138 ms, sys: 6.99 ms, total: 145 ms
Wall time: 440 ms

[26]:

%%time
 
mean_delay_res, std_delay_res = dask.compute(mean_delay, std_delay)

CPU times: user 90.9 ms, sys: 3.84 ms, total: 94.8 ms
Wall time: 270 ms

[27]:

dask.visualize(mean_delay, std_delay)

[27]:

MemoryError:  ...

[28]:

crs_dep_time = df.CRSDepTime.head(10)
crs_dep_time

[28]:

0    1540
1    1540
2    1540
3    1540
4    1540
5    1540
6    1540
7    1540
8    1540
9    1540
Name: CRSDepTime, dtype: int64

[29]:

import pandas as pd
 
# Get the first 10 dates to complement our `crs_dep_time`
date = df.Date.head(10)
 
# Get hours as an integer, convert to a timedelta
hours = crs_dep_time // 100
hours_timedelta = pd.to_timedelta(hours, unit='h')
 
# Get minutes as an integer, convert to a timedelta
minutes = crs_dep_time % 100
minutes_timedelta = pd.to_timedelta(minutes, unit='m')
 
# Apply the timedeltas to offset the dates by the departure time
departure_timestamp = date + hours_timedelta + minutes_timedelta
departure_timestamp

[29]:

0   1990-01-01 15:40:00
1   1990-01-02 15:40:00
2   1990-01-03 15:40:00
3   1990-01-04 15:40:00
4   1990-01-05 15:40:00
5   1990-01-06 15:40:00
6   1990-01-07 15:40:00
7   1990-01-08 15:40:00
8   1990-01-09 15:40:00
9   1990-01-10 15:40:00
dtype: datetime64[ns]

[30]:

# Look at the docs for `map_partitions`
 
help(df.CRSDepTime.map_partitions)

Help on method map_partitions in module dask.dataframe.core:
 
map_partitions(func, *args, **kwargs) method of dask.dataframe.core.Series instance
    Apply Python function on each DataFrame partition.
 
    Note that the index and divisions are assumed to remain unchanged.
 
    Parameters
    ----------
    func : function
        Function applied to each partition.
    args, kwargs :
        Arguments and keywords to pass to the function. The partition will
        be the first argument, and these will be passed *after*. Arguments
        and keywords may contain ``Scalar``, ``Delayed`` or regular
        python objects. DataFrame-like args (both dask and pandas) will be
        repartitioned to align (if necessary) before applying the function.
    meta : pd.DataFrame, pd.Series, dict, iterable, tuple, optional
        An empty ``pd.DataFrame`` or ``pd.Series`` that matches the dtypes
        and column names of the output. This metadata is necessary for
        many algorithms in dask dataframe to work.  For ease of use, some
        alternative inputs are also available. Instead of a ``DataFrame``,
        a ``dict`` of ``{name: dtype}`` or iterable of ``(name, dtype)``
        can be provided (note that the order of the names should match the
        order of the columns). Instead of a series, a tuple of ``(name,
        dtype)`` can be used. If not provided, dask will try to infer the
        metadata. This may lead to unexpected results, so providing
        ``meta`` is recommended. For more information, see
        ``dask.dataframe.utils.make_meta``.
 
    Examples
    --------
    Given a DataFrame, Series, or Index, such as:
 
    >>> import dask.dataframe as dd
    >>> df = pd.DataFrame({'x': [1, 2, 3, 4, 5],
    ...                    'y': [1., 2., 3., 4., 5.]})
    >>> ddf = dd.from_pandas(df, npartitions=2)
 
    One can use ``map_partitions`` to apply a function on each partition.
    Extra arguments and keywords can optionally be provided, and will be
    passed to the function after the partition.
 
    Here we apply a function with arguments and keywords to a DataFrame,
    resulting in a Series:
 
    >>> def myadd(df, a, b=1):
    ...     return df.x + df.y + a + b
    >>> res = ddf.map_partitions(myadd, 1, b=2)
    >>> res.dtype
    dtype('float64')
 
    By default, dask tries to infer the output metadata by running your
    provided function on some fake data. This works well in many cases, but
    can sometimes be expensive, or even fail. To avoid this, you can
    manually specify the output metadata with the ``meta`` keyword. This
    can be specified in many forms, for more information see
    ``dask.dataframe.utils.make_meta``.
 
    Here we specify the output is a Series with no name, and dtype
    ``float64``:
 
    >>> res = ddf.map_partitions(myadd, 1, b=2, meta=(None, 'f8'))
 
    Here we map a function that takes in a DataFrame, and returns a
    DataFrame with a new column:
 
    >>> res = ddf.map_partitions(lambda df: df.assign(z=df.x * df.y))
    >>> res.dtypes
    x      int64
    y    float64
    z    float64
    dtype: object
 
    As before, the output metadata can also be specified manually. This
    time we pass in a ``dict``, as the output is a DataFrame:
 
    >>> res = ddf.map_partitions(lambda df: df.assign(z=df.x * df.y),
    ...                          meta={'x': 'i8', 'y': 'f8', 'z': 'f8'})
 
    In the case where the metadata doesn't change, you can also pass in
    the object itself directly:
 
    >>> res = ddf.map_partitions(lambda df: df.head(), meta=df)
 
    Also note that the index and divisions are assumed to remain unchanged.
    If the function you're mapping changes the index/divisions, you'll need
    to clear them afterwards:
 
    >>> ddf.map_partitions(func).clear_divisions()  # doctest: +SKIP
 

[31]:

hours = df.CRSDepTime // 100
# hours_timedelta = pd.to_timedelta(hours, unit='h')
hours_timedelta = hours.map_partitions(pd.to_timedelta, unit='h')
 
minutes = df.CRSDepTime % 100
# minutes_timedelta = pd.to_timedelta(minutes, unit='m')
minutes_timedelta = minutes.map_partitions(pd.to_timedelta, unit='m')
 
departure_timestamp = df.Date + hours_timedelta + minutes_timedelta

[32]:

departure_timestamp

[32]:

Dask Series Structure:
npartitions=10
    datetime64[ns]
               ...
         ...
               ...
               ...
dtype: datetime64[ns]
Dask Name: add, 110 tasks

[33]:

departure_timestamp.head()

[33]:

0   1990-01-01 15:40:00
1   1990-01-02 15:40:00
2   1990-01-03 15:40:00
3   1990-01-04 15:40:00
4   1990-01-05 15:40:00
dtype: datetime64[ns]

[34]:

def compute_departure_timestamp(df):
    pass  # TODO: implement this

[35]:

departure_timestamp = df.map_partitions(compute_departure_timestamp)
 
departure_timestamp.head()

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-35-b8c8b4861a84> in <module>
      1 departure_timestamp = df.map_partitions(compute_departure_timestamp)
      2
----> 3 departure_timestamp.head()
 
~/miniconda/envs/test/lib/python3.7/site-packages/dask/dataframe/core.py in head(self, n, npartitions, compute)
    948             Whether to compute the result, default is True.
    949         """
--> 950         return self._head(n=n, npartitions=npartitions, compute=compute, safe=True)
    951
    952     def _head(self, n, npartitions, compute, safe):
 
~/miniconda/envs/test/lib/python3.7/site-packages/dask/dataframe/core.py in _head(self, n, npartitions, compute, safe)
    981
    982         if compute:
--> 983             result = result.compute()
    984         return result
    985
 
~/miniconda/envs/test/lib/python3.7/site-packages/dask/base.py in compute(self, **kwargs)
    163         dask.base.compute
    164         """
--> 165         (result,) = compute(self, traverse=False, **kwargs)
    166         return result
    167
 
~/miniconda/envs/test/lib/python3.7/site-packages/dask/base.py in compute(*args, **kwargs)
    434     keys = [x.__dask_keys__() for x in collections]
    435     postcomputes = [x.__dask_postcompute__() for x in collections]
--> 436     results = schedule(dsk, keys, **kwargs)
    437     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
    438
 
~/miniconda/envs/test/lib/python3.7/site-packages/distributed/client.py in get(self, dsk, keys, restrictions, loose_restrictions, resources, sync, asynchronous, direct, retries, priority, fifo_timeout, actors, **kwargs)
   2570                     should_rejoin = False
   2571             try:
-> 2572                 results = self.gather(packed, asynchronous=asynchronous, direct=direct)
   2573             finally:
   2574                 for f in futures.values():
 
~/miniconda/envs/test/lib/python3.7/site-packages/distributed/client.py in gather(self, futures, errors, direct, asynchronous)
   1870                 direct=direct,
   1871                 local_worker=local_worker,
-> 1872                 asynchronous=asynchronous,
   1873             )
   1874
 
~/miniconda/envs/test/lib/python3.7/site-packages/distributed/client.py in sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
    765         else:
    766             return sync(
--> 767                 self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
    768             )
    769
 
~/miniconda/envs/test/lib/python3.7/site-packages/distributed/utils.py in sync(loop, func, callback_timeout, *args, **kwargs)
    332     if error[0]:
    333         typ, exc, tb = error[0]
--> 334         raise exc.with_traceback(tb)
    335     else:
    336         return result[0]
 
~/miniconda/envs/test/lib/python3.7/site-packages/distributed/utils.py in f()
    316             if callback_timeout is not None:
    317                 future = gen.with_timeout(timedelta(seconds=callback_timeout), future)
--> 318             result[0] = yield future
    319         except Exception as exc:
    320             error[0] = sys.exc_info()
 
~/miniconda/envs/test/lib/python3.7/site-packages/tornado/gen.py in run(self)
    733
    734                     try:
--> 735                         value = future.result()
    736                     except Exception:
    737                         exc_info = sys.exc_info()
 
~/miniconda/envs/test/lib/python3.7/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
   1726                             exc = CancelledError(key)
   1727                         else:
-> 1728                             raise exception.with_traceback(traceback)
   1729                         raise exc
   1730                     if errors == "skip":
 
~/miniconda/envs/test/lib/python3.7/site-packages/dask/dataframe/core.py in safe_head()
   5739
   5740 def safe_head(df, n):
-> 5741     r = M.head(df, n)
   5742     if len(r) != n:
   5743         msg = (
 
~/miniconda/envs/test/lib/python3.7/site-packages/dask/utils.py in __call__()
    879
    880     def __call__(self, obj, *args, **kwargs):
--> 881         return getattr(obj, self.method)(*args, **kwargs)
    882
    883     def __reduce__(self):
 
AttributeError: 'NoneType' object has no attribute 'head'

[36]:

def compute_departure_timestamp(df):
    hours = df.CRSDepTime // 100
    hours_timedelta = pd.to_timedelta(hours, unit='h')
 
    minutes = df.CRSDepTime % 100
    minutes_timedelta = pd.to_timedelta(minutes, unit='m')
 
    return df.Date + hours_timedelta + minutes_timedelta
 
departure_timestamp = df.map_partitions(compute_departure_timestamp)
departure_timestamp.head()

[36]:

0   1990-01-01 15:40:00
1   1990-01-02 15:40:00
2   1990-01-03 15:40:00
3   1990-01-04 15:40:00
4   1990-01-05 15:40:00
dtype: datetime64[ns]

[37]:

client.shutdown()