Looks and feels like the pandas API, but for parallel and distributed workflows.
At its core, the dask.dataframe
module implements a "blocked parallel" DataFrame
object that looks and feels like the pandas API, but for parallel and distributed workflows. One Dask DataFrame
is comprised of many in-memory pandas DataFrame
s 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
¶pandas is great for tabular datasets that fit in memory. A general rule of thumb for pandas is:
"Have 5 to 10 times as much RAM as the size of your dataset"
~ Wes McKinney (2017) in 10 things I hate about pandas
Here "size of dataset" means dataset size on the disk.
Dask becomes useful when the datasets exceed the above rule.
In this notebook, you will be working with the New York City Airline data. This dataset is only ~200MB, so that you can download it in a reasonable time, but dask.dataframe
will scale to datasets much larger than memory.
Create the datasets you will be using in this notebook:
%run prep.py -d flights
Create a local Dask cluster and connect it to the client. Don't worry about this bit of code for now, you will learn more in the Distributed notebook.
from dask.distributed import Client
client = Client(n_workers=4)
client
Dask Distributed provides a useful Dashboard to visualize the state of your cluster and computations.
If you're on JupyterLab or Binder, you can use the Dask JupyterLab extension (which should be already installed in your environment) to open the dashboard plots:
Alternatively, click on the dashboard link displayed in the Client details above: http://127.0.0.1:8787/status. It will open a new browser tab with the Dashboard.
Let's read 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.
import os
import dask
By convention, we import the module dask.dataframe
as dd
, and call the corresponding DataFrame
object ddf
.
Note: The term "Dask DataFrame" is slightly overloaded. Depending on the context, it can refer to the module or the DataFrame object. To avoid confusion, throughout this notebook:
dask.dataframe
(note the all lowercase) refers to the API, andDataFrame
(note the CamelCase) refers to the object.The following filename includes a glob pattern *
, so all files in the path matching that pattern will be read into the same DataFrame
.
import dask.dataframe as dd
ddf = dd.read_csv(
os.path.join("data", "nycflights", "*.csv"), parse_dates={"Date": [0, 1, 2]}
)
ddf
Dask has not loaded the data yet, it has:
Notice that the representation 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.
Most Dask Collections, including Dask DataFrame
are evaluated lazily, which means Dask constructs the logic (called task graph) of your computation immediately but "evaluates" them only when necessary. You can view this task graph using .visualize()
.
You will learn more about this in the Delayed notebook, but for now, note that we need to call .compute()
to trigger actual computations.
ddf.visualize()
Some functions like len
and head
also trigger a computation. Specifically, calling len
will:
# load and count number of rows
len(ddf)
You can view the start and end of the data as you would in pandas:
ddf.head()
ddf.tail()
# 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`.
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.
ddf = dd.read_csv(
os.path.join("data", "nycflights", "*.csv"),
parse_dates={"Date": [0, 1, 2]},
dtype={"TailNum": str, "CRSElapsedTime": float, "Cancelled": bool},
)
ddf.tail() # now works
If you're thinking about distributed computing, your data is probably stored remotely on services (like Amazon's S3 or Google's cloud storage) and is in a friendlier format (like Parquet). Dask can read data in various formats directly from these remote locations lazily and in parallel.
Here's how you can read the NYC taxi cab data from Amazon S3:
ddf = dd.read_parquet(
"s3://nyc-tlc/trip data/yellow_tripdata_2012-*.parquet",
)
You can also leverage Parquet-specific optimizations like column selection and metadata handling, learn more in the Dask documentation on working with Parquet files.
dask.dataframe
¶Let's compute the maximum of the flight delay.
With just pandas, we would loop over each file to find the individual maximums, then find the final maximum over all the individual maximums.
import pandas as pd
files = os.listdir(os.path.join('data', 'nycflights'))
maxes = []
for file in files:
df = pd.read_csv(os.path.join('data', 'nycflights', file))
maxes.append(df.DepDelay.max())
final_max = max(maxes)
dask.dataframe
lets us write pandas-like code, that operates on larger-than-memory datasets in parallel.
%%time
result = ddf.DepDelay.max()
result.compute()
This creates the lazy computation for us and then runs it.
Note: Dask will delete intermediate results (like the full pandas DataFrame for each file) as soon as possible. This means you can handle datasets that are larger than memory but, 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?)
You can view the underlying task graph using .visualize()
:
# notice the parallelism
result.visualize()
In this section you will 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()
.
Hint: how would you check how many items are in a list?
# Your code here
len(ddf)
Hint: use boolean indexing.
# Your code here
len(ddf[~ddf.Cancelled])
# Your code here
ddf[~ddf.Cancelled].groupby("Origin").Origin.count().compute()
# Your code here
ddf.groupby("Origin").DepDelay.mean().compute()
# Your code here
ddf.groupby("DayOfWeek").DepDelay.mean().idxmax().compute()
# Your code here
ddf["Distance"].apply(
lambda x: x + 1
).compute() # don't worry about the warning, we'll discuss in the next sections
# OR
(ddf["Distance"] + 1).compute()
When computing all of the above, we sometimes did the same operation more than once. For most operations, dask.dataframe
stores the arguments, allowing duplicate computations to be shared and only computed once.
For example, let's 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 steps required to get the result.
If you compute them with two calls to compute, there is no sharing of intermediate computations.
non_canceled = ddf[~ddf.Cancelled]
mean_delay = non_canceled.DepDelay.mean()
std_delay = non_canceled.DepDelay.std()
%%time
mean_delay_res = mean_delay.compute()
std_delay_res = std_delay.compute()
dask.compute
¶But let's 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 (you might want to use filename='graph.pdf'
to save the graph to disk so that you can zoom in more easily):
dask.visualize(mean_delay, std_delay, engine="cytoscape")
.persist()
¶While using a distributed scheduler (you will learn more about schedulers in the upcoming notebooks), you can keep some data that you want to use often in the distributed memory.
persist
generates "Futures" (more on this later as well) and stores them in the same structure as your output. You can use persist
with any data or computation that fits in memory.
If you want to analyze data only for non-canceled flights departing from JFK airport, you can either have two compute calls like in the previous section:
non_cancelled = ddf[~ddf.Cancelled]
ddf_jfk = non_cancelled[non_cancelled.Origin == "JFK"]
%%time
ddf_jfk.DepDelay.mean().compute()
ddf_jfk.DepDelay.sum().compute()
Or, consider persisting that subset of data in memory.
See the "Graph" dashboard plot, the red squares indicate persisted data stored as Futures in memory. You will also notice an increase in Worker Memory (another dashboard plot) consumption.
ddf_jfk = ddf_jfk.persist() # returns back control immediately
%%time
ddf_jfk.DepDelay.mean().compute()
ddf_jfk.DepDelay.std().compute()
Analyses on this persisted data is faster because we are not repeating the loading and selecting (non-canceled, JFK departure) operations.
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.
What if you want to use some custom functions that aren't (or can't be) implemented for Dask DataFrame yet?
You can open an issue on the Dask issue tracker to check how feasible the function could be to implement, and you can consider contributing this function to Dask.
In case it's a custom function or tricky to implement, dask.dataframe
provides a few methods to make applying custom functions to Dask DataFrames easier:
map_partitions
: to run a function on each partition (each pandas DataFrame) of the Dask DataFramemap_overlap
: to run a function on each partition (each pandas DataFrame) of the Dask DataFrame, with some rows shared between neighboring partitionsreduction
: for custom row-wise reduction operations.Let's take a quick look at the map_partitions()
function:
help(ddf.map_partitions)
The "Distance" column in ddf
is currently in miles. Let's say we want to convert the units to kilometers and we have a general helper function as shown below. In this case, we can use map_partitions
to apply this function across each of the internal pandas DataFrame
s in parallel.
def my_custom_converter(df, multiplier=1):
return df * multiplier
meta = pd.Series(name="Distance", dtype="float64")
distance_km = ddf.Distance.map_partitions(
my_custom_converter, multiplier=0.6, meta=meta
)
distance_km.visualize()
distance_km.head()
meta
?¶Since Dask operates lazily, it doesn't always have enough information to infer the output structure (which includes datatypes) of certain operations.
meta
is a suggestion to Dask about the output of your computation. Importantly, meta
never infers with the output structure. Dask uses this meta
until it can determine the actual output structure.
Even though there are many ways to define meta
, we suggest using a small pandas Series or DataFrame that matches the structure of your final output.
It's good practice to always close any Dask cluster you create:
client.shutdown()