04_dataframe.ipynb

Kernel: Python 3

Dask DataFrames

We finished Chapter 02 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 `DataFrame`s in a way that is mindful of potential parallelism and memory constraints.

Related Documentation

• Dask DataFrame documentation

• Pandas documentation

Main Take-aways

1. Dask.dataframe should be familiar to Pandas users

2. The partitioning of dataframes is important for efficient queries

Setup

We create artifical data.

from prep import accounts_csvs
accounts_csvs(3, 1000000, 500)

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

This works just like pandas.read_csv, except on multiple csv files at once.

filename

import dask.dataframe as dd
df = dd.read_csv(filename)
# load and count number of rows
df.head()

len(df)

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.

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

df

We can view the start and end of the data

df.head()

df.tail()  # this fails

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.

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

df.tail()  # now works

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.

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

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

Some things to note:

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

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

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

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?

# Your code here

%load solutions/03-dask-dataframe-rows.py

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

With pandas, you would use boolean indexing.

# Your code here

%load solutions/03-dask-dataframe-non-cancelled.py

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

Hint: use df.groupby.

# Your code here

%load solutions/03-dask-dataframe-non-cancelled-per-airport.py

4.) What was the average departure delay from each airport?

Note, this is the same computation you did in the previous notebook (is this approach faster or slower?)

# Your code here

df.columns

%load solutions/03-dask-dataframe-delay-per-airport.py

5.) What day of the week has the worst average departure delay?

# Your code here

%load solutions/03-dask-dataframe-delay-per-day.py

Sharing Intermediate Results

When computing all of the above, we sometimes did the same operation more than once. For most operations, dask.dataframe hashes the arguments, allowing duplicate computations to be shared, and only computed once.

For example, lets compute the mean and standard deviation for departure delay of all non-canceled flights. Since dask operations are lazy, those values aren't the final results yet. They're just the recipe required to get the result.

If we compute them with two calls to compute, there is no sharing of intermediate computations.

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

%%time

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

But lets try by passing both to a single compute call.

%%time

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

Using dask.compute takes roughly 1/2 the time. This is because the task graphs for both results are merged when calling dask.compute, allowing shared operations to only be done once instead of twice. In particular, using dask.compute only does the following once:

• the calls to read_csv

• the filter (df[~df.Cancelled])

• some of the necessary reductions (sum, count)

To see what the merged task graphs between multiple results look like (and what's shared), you can use the dask.visualize function (we might want to use filename='graph.pdf' to zoom in on the graph better):

dask.visualize(mean_delay, std_delay)

How does this compare to Pandas?

Pandas is more mature and fully featured than dask.dataframe. If your data fits in memory then you should use Pandas. The dask.dataframe module gives you a limited pandas experience when you operate on datasets that don't fit comfortably in memory.

During this tutorial we provide a small dataset consisting of a few CSV files. This dataset is 45MB on disk that expands to about 400MB in memory (the difference is caused by using object dtype for strings). This dataset is small enough that you would normally use Pandas.

We've chosen this size so that exercises finish quickly. Dask.dataframe only really becomes meaningful for problems significantly larger than this, when Pandas breaks with the dreaded

MemoryError:  ...

Furthermore, the distributed scheduler allows the same dataframe expressions to be executed across a cluster. To enable massive "big data" processing, one could execute data ingestion functions such as read_csv, where the data is held on storage accessible to every worker node (e.g., amazon's S3), and because most operations begin by selecting only some columns, transforming and filtering the data, only relatively small amounts of data need to be communicated between the machines.

Dask.dataframe operations use pandas operations internally. Generally they run at about the same speed except in the following two cases:

1. Dask introduces a bit of overhead, around 1ms per task. This is usually negligible.

2. When Pandas releases the GIL (coming to groupby in the next version) dask.dataframe can call several pandas operations in parallel within a process, increasing speed somewhat proportional to the number of cores. For operations which don't release the GIL, multiple processes would be needed to get the same speedup.

Dask DataFrame Data Model

For the most part, a Dask DataFrame feels like a pandas DataFrame. So far, the biggest difference we've seen is that Dask operations are lazy; they build up a task graph instead of executing immediately (more details coming in Schedulers). This lets Dask do operations in parallel and out of core.

In Dask Arrays, we saw that a dask.array was composed of many NumPy arrays, chunked along one or more dimensions. It's similar for dask.dataframe: a Dask DataFrame is composed of many pandas DataFrames. For dask.dataframe the chunking happens only along the index.

We call each chunk a partition, and the upper / lower bounds are divisions. Dask can store information about the divisions. For now, partitions come up when you write custom functions to apply to Dask DataFrames

Converting CRSDepTime to a timestamp

This dataset stores timestamps as HHMM, which are read in as integers in read_csv:

crs_dep_time = df.CRSDepTime.head(10)
crs_dep_time

To convert these to timestamps of scheduled departure time, we need to convert these integers into pd.Timedelta objects, and then combine them with the Date column.

In pandas we'd do this using the pd.to_timedelta function, and a bit of arithmetic:

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

Custom code and Dask Dataframe

We could swap out pd.to_timedelta for dd.to_timedelta and do the same operations on the entire dask DataFrame. But let's say that Dask hadn't implemented a dd.to_timedelta that works on Dask DataFrames. What would you do then?

dask.dataframe provides a few methods to make applying custom functions to Dask DataFrames easier:

• map_partitions

• map_overlap

• reduction

Here we'll just be discussing map_partitions, which we can use to implement to_timedelta on our own:

# Look at the docs for `map_partitions`

help(df.CRSDepTime.map_partitions)

The basic idea is to apply a function that operates on a DataFrame to each partition. In this case, we'll apply pd.to_timedelta.

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

departure_timestamp

departure_timestamp.head()

Exercise: Rewrite above to use a single call to map_partitions

This will be slightly more efficient than two separate calls, as it reduces the number of tasks in the graph.

def compute_departure_timestamp(df):
    # TODO

departure_timestamp = df.map_partitions(compute_departure_timestamp)

departure_timestamp.head()

%load solutions/03-dask-dataframe-map-partitions.py

Limitations

What doesn't work?

Dask.dataframe only covers a small but well-used portion of the Pandas API. This limitation is for two reasons:

1. The Pandas API is huge

2. Some operations are genuinely hard to do in parallel (e.g. sort)

Additionally, some important operations like set_index work, but are slower than in Pandas because they include substantial shuffling of data, and may write out to disk.

What definitely works?

• Trivially parallelizable operations (fast):

  • Elementwise operations: df.x + df.y

  • Row-wise selections: df[df.x > 0]

  • Loc: df.loc[4.0:10.5]

  • Common aggregations: df.x.max()

  • Is in: df[df.x.isin([1, 2, 3])]

  • Datetime/string accessors: df.timestamp.month

• Cleverly parallelizable operations (also fast):

  • groupby-aggregate (with common aggregations): df.groupby(df.x).y.max()

  • value_counts: df.x.value_counts

  • Drop duplicates: df.x.drop_duplicates()

  • Join on index: dd.merge(df1, df2, left_index=True, right_index=True)

• Operations requiring a shuffle (slow-ish, unless on index)

  • Set index: df.set_index(df.x)

  • groupby-apply (with anything): df.groupby(df.x).apply(myfunc)

  • Join not on the index: pd.merge(df1, df2, on='name')

• Ingest operations

  • Files: dd.read_csv, dd.read_parquet, dd.read_json, dd.read_orc, etc.

  • Pandas: dd.from_pandas

  • Anything supporting numpy slicing: dd.from_array

  • From any set of functions creating sub dataframes via dd.from_delayed.

  • Dask.bag: mybag.to_dataframe(columns=[...])