In the 01-data-access example we show how Dask Dataframes can read and store data in many of the same formats as Pandas dataframes. One key difference, when using Dask Dataframes is that instead of opening a single file with a function like pandas.read_csv, we typically open many files at once with dask.dataframe.read_csv. This enables us to treat a collection of files as a single dataset. Most of the time this works really well. But real data is messy and in this notebook we will explore a more advanced technique to bring messy datasets into a dask dataframe.
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.
from dask.distributed import Client
client = Client(n_workers=1, threads_per_worker=4, processes=True, memory_limit='2GB')
client
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.
import dask
df = dask.datasets.timeseries()
df
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, index=False);
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.
!ls data/*.csv | head
import dask.dataframe as dd
df = dd.read_csv('data/2000-*-*.csv')
df
df.head()
Let's look at some statistics on the data
df.describe().compute()
Now this works great, and in most cases dd.read_csv or dd.read_parquet etc are the preferred way to read in large collections of data files into a dask dataframe, but real world data is often very messy and some files may be broken or badly formatted. To simulate this we are going to create some fake messy data by tweaking our example csv files. For the file data/2000-01-05.csv
we will replace with no data and for the file data/2000-01-07.csv
we will remove the y
column
# corrupt the data in data/2000-01-05.csv
with open('data/2000-01-05.csv', 'w') as f:
f.write('')
# remove y column from data/2000-01-07.csv
import pandas as pd
df = pd.read_csv('data/2000-01-07.csv')
del df['y']
df.to_csv('data/2000-01-07.csv', index=False)
!head data/2000-01-05.csv
!head data/2000-01-07.csv
Let's try reading in the collection of files again
df = dd.read_csv('data/2000-*-*.csv')
df.head()
Ok this looks like it worked, let us calculate the dataset statistics again
df.describe().compute()
So what happened?
When creating a dask dataframe from a collection of files, dd.read_csv samples the first few files in the dataset to determine the datatypes and columns available. Since it has not opened all the files it does not now if some of them are corrupt. Hence, df.head()
works since it is only looking at the first file. df.describe.compute()
fails because of the corrupt data in data/2000-01-05.csv
To get around this problem we are going to use a more advanced technique to build our dask dataframe. This method can also be used any time some custom logic is required when reading each file. Essentially, we are going to build a function that uses pandas and some error checking and returns a pandas dataframe. If we find a bad data file we will either find a way to fix/clean the data or we will return and empty pandas dataframe with the same structure as the good data.
import numpy as np
import io
def read_data(filename):
# for this to work we need to explicitly set the datatypes of our pandas dataframe
dtypes = {'id': int, 'name': str, 'x': float, 'y': float}
try:
# try reading in the data with pandas
df = pd.read_csv(filename, dtype=dtypes)
except:
# if this fails create an empty pandas dataframe with the same dtypes as the good data
df = pd.read_csv(io.StringIO(''), names=dtypes.keys(), dtype=dtypes)
# for the case with the missing column, add a column of data with NaN's
if 'y' not in df.columns:
df['y'] = np.NaN
return df
Let's test this function on a good file and the two bad files
# test function on a normal file
read_data('data/2000-01-01.csv').head()
# test function on the empty file
read_data('data/2000-01-05.csv').head()
# test function on the file missing the y column
read_data('data/2000-01-07.csv').head()
First we take our read_data
function and convert it to a dask delayed function
from dask import delayed
read_data = delayed(read_data)
Let us look at what the function does now
df = read_data('data/2000-01-01.csv')
df
It creates a delayed object, to actually run read the file we need to run .compute()
df.compute()
Now let's build a list of all the available csv files
# loop over all the files
from glob import glob
files = glob('data/2000-*-*.csv')
files
Now we run the delayed read_data function on each file in the list
df = [read_data(file) for file in files]
df
Then we use dask.dataframe.from_delayed. This function creates a Dask DataFrame from a list of delayed objects as long as each delayed object returns a pandas dataframe. The structure of each individual dataframe returned must also be the same.
df = dd.from_delayed(df, meta={'id': int, 'name': str, 'x': float, 'y': float})
df
Note: we provided the dtypes in the meta
keyword to explicitly tell Dask Dataframe what kind of dataframe to expect. If we did not do this Dask would infer this from the first delayed object which could be slow if it was a large csv file
df.head()
df.describe().compute()
To recap, in this example, we looked at an approach to create a Dask Dataframe from a collection of many data files. Typically you would use built-in functions like dd.read_csv
or dd.read_parquet
to do this. Sometimes, this is not possible because of messy/corrupted files in your dataset or some custom processing that might need to be done.
In these cases, you can build a Dask DataFrame with the following steps.
dask.delayed
functiondd.from_delayed
to covert the list of delayed objects to a Dask DataframeThis same technique can be used in other situations as well. Another example might be data files that require using a specialized reader, or several transformations before they can be converted to a pandas dataframe.