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.
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
Main Take-aways
%run prep.py -d flights
from dask.distributed import Client
client = Client(n_workers=4)
We create artifical data.
from prep import accounts_csvs
accounts_csvs()
import os
import dask
filename = os.path.join('data', 'accounts.*.csv')
filename
Filename includes a glob pattern *
, so all files in the path matching that pattern will be read into the same Dask DataFrame.
import dask.dataframe as dd
df = dd.read_csv(filename)
df.head()
# load and count number of rows
len(df)
What happened here?
len()
applied to itLets 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 dtypes.
df
We can view the start and end of the data
df.head()
df.tail() # this fails
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:
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.sample
keyword (in bytes)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
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:
dask.delayed
, we need to call .compute()
when we're done. Up until this point everything is lazy.As with Delayed
objects, you can view the underlying task graph using the .visualize
method:
# notice the parallelism
df.DepDelay.max().visualize()
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
.
If you aren't familiar with pandas, how would you check how many records are in a list of tuples?
# Your code here
len(df)
With pandas, you would use boolean indexing.
# Your code here
len(df[~df.Cancelled])
Hint: use df.groupby
.
# Your code here
df[~df.Cancelled].groupby('Origin').Origin.count().compute()
Note, this is the same computation you did in the previous notebook (is this approach faster or slower?)
# Your code here
df.groupby("Origin").DepDelay.mean().compute()
# Your code here
df.groupby("DayOfWeek").DepDelay.mean().compute()
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:
read_csv
df[~df.Cancelled]
)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)
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. 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:
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.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
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
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:
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()
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):
pass # TODO: implement this
departure_timestamp = df.map_partitions(compute_departure_timestamp)
departure_timestamp.head()
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()
Dask.dataframe only covers a small but well-used portion of the Pandas API. This limitation is for two reasons:
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.
client.shutdown()