You can run this notebook in a live session 
or view it on Github.
Dask Bags¶
Dask Bag implements operations like map, filter, groupby and
aggregations on collections of Python objects. It does this in parallel
and in small memory using Python iterators. It is similar to a parallel
version of itertools or a Pythonic version of the PySpark RDD.
Dask Bags are often used to do simple preprocessing on log files, JSON records, or other user defined Python objects.
Full API documentation is available here: http://docs.dask.org/en/latest/bag-api.html
Start Dask Client for Dashboard¶
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.
In [1]:
from dask.distributed import Client, progress
client = Client(n_workers=4, threads_per_worker=1)
client
Out[1]:
Client
|
Cluster
|
Create Random Data¶
We create a random set of record data and store it to disk as many JSON files. This will serve as our data for this notebook.
In [2]:
import dask
import json
import os
os.makedirs('data', exist_ok=True) # Create data/ directory
b = dask.datasets.make_people() # Make records of people
b.map(json.dumps).to_textfiles('data/*.json') # Encode as JSON, write to disk
Out[2]:
['data/0.json',
'data/1.json',
'data/2.json',
'data/3.json',
'data/4.json',
'data/5.json',
'data/6.json',
'data/7.json',
'data/8.json',
'data/9.json']
Read JSON data¶
Now that we have some JSON data in a file lets take a look at it with Dask Bag and Python JSON module.
In [3]:
!head data/0.json -n 2
{"age": 29, "name": ["Preston", "Ray"], "occupation": "Haulage Contractor", "telephone": "113.919.6933", "address": {"address": "438 Southwood Spur", "city": "St. Charles"}, "credit-card": {"number": "2471 7972 2251 8059", "expiration-date": "06/21"}}
{"age": 57, "name": ["Beverlee", "Price"], "occupation": "Service Engineer", "telephone": "045-216-5761", "address": {"address": "326 San Bruno Stravenue", "city": "Coronado"}, "credit-card": {"number": "3406 240769 42065", "expiration-date": "09/25"}}
In [4]:
import dask.bag as db
import json
b = db.read_text('data/*.json').map(json.loads)
b
Out[4]:
dask.bag<map-loa..., npartitions=10>
In [5]:
b.take(2)
Out[5]:
({'age': 29,
'name': ['Preston', 'Ray'],
'occupation': 'Haulage Contractor',
'telephone': '113.919.6933',
'address': {'address': '438 Southwood Spur', 'city': 'St. Charles'},
'credit-card': {'number': '2471 7972 2251 8059',
'expiration-date': '06/21'}},
{'age': 57,
'name': ['Beverlee', 'Price'],
'occupation': 'Service Engineer',
'telephone': '045-216-5761',
'address': {'address': '326 San Bruno Stravenue', 'city': 'Coronado'},
'credit-card': {'number': '3406 240769 42065', 'expiration-date': '09/25'}})
Map, Filter, Aggregate¶
We can process this data by filtering out only certain records of interest, mapping functions over it to process our data, and aggregating those results to a total value.
In [6]:
b.filter(lambda record: record['age'] > 30).take(2) # Select only people over 30
Out[6]:
({'age': 57,
'name': ['Beverlee', 'Price'],
'occupation': 'Service Engineer',
'telephone': '045-216-5761',
'address': {'address': '326 San Bruno Stravenue', 'city': 'Coronado'},
'credit-card': {'number': '3406 240769 42065', 'expiration-date': '09/25'}},
{'age': 65,
'name': ['Shery', 'Bradley'],
'occupation': 'Bus Company',
'telephone': '+1-(919)-146-4994',
'address': {'address': '253 Walkway Avenue', 'city': 'Rockwall'},
'credit-card': {'number': '2671 6820 1040 3714',
'expiration-date': '10/17'}})
In [7]:
b.map(lambda record: record['occupation']).take(2) # Select the occupation field
Out[7]:
('Haulage Contractor', 'Service Engineer')
In [8]:
b.count().compute() # Count total number of records
Out[8]:
10000
Chain computations¶
It is common to do many of these steps in one pipeline, only calling
compute or take at the end.
In [9]:
result = (b.filter(lambda record: record['age'] > 30)
.map(lambda record: record['occupation'])
.frequencies(sort=True)
.topk(10, key=1))
result
Out[9]:
dask.bag<topk-ag..., npartitions=1>
As with all lazy Dask collections, we need to call compute to
actually evaluate our result. The take method used in earlier
examples is also like compute and will also trigger computation.
In [10]:
result.compute()
Out[10]:
[('Advertising Executive', 14),
('Heating Engineer', 14),
('Quantity Surveyor', 14),
('Labelling Operator', 14),
('Wheel Clamper', 13),
('Parts Supervisor', 13),
('Hospital Worker', 13),
('Barmaid', 13),
('Screen Writer', 13),
('Landowner', 13)]
Transform and Store¶
Sometimes we want to compute aggregations as above, but sometimes we
want to store results to disk for future analyses. For that we can use
methods like to_textfiles and json.dumps, or we can convert to
Dask Dataframes and use their storage systems, which we’ll see more of
in the next section.
In [11]:
(b.filter(lambda record: record['age'] > 30) # Select records of interest
.map(json.dumps) # Convert Python objects to text
.to_textfiles('data/processed.*.json')) # Write to local disk
Out[11]:
['data/processed.0.json',
'data/processed.1.json',
'data/processed.2.json',
'data/processed.3.json',
'data/processed.4.json',
'data/processed.5.json',
'data/processed.6.json',
'data/processed.7.json',
'data/processed.8.json',
'data/processed.9.json']
Convert to Dask Dataframes¶
Dask Bags are good for reading in initial data, doing a bit of pre-processing, and then handing off to some other more efficient form like Dask Dataframes. Dask Dataframes use Pandas internally, and so can be much faster on numeric data and also have more complex algorithms.
However, Dask Dataframes also expect data that is organized as flat columns. It does not support nested JSON data very well (Bag is better for this).
Here we make a function to flatten down our nested data structure, map that across our records, and then convert that to a Dask Dataframe.
In [12]:
b.take(1)
Out[12]:
({'age': 29,
'name': ['Preston', 'Ray'],
'occupation': 'Haulage Contractor',
'telephone': '113.919.6933',
'address': {'address': '438 Southwood Spur', 'city': 'St. Charles'},
'credit-card': {'number': '2471 7972 2251 8059',
'expiration-date': '06/21'}},)
In [13]:
def flatten(record):
return {
'age': record['age'],
'occupation': record['occupation'],
'telephone': record['telephone'],
'credit-card-number': record['credit-card']['number'],
'credit-card-expiration': record['credit-card']['expiration-date'],
'name': ' '.join(record['name']),
'street-address': record['address']['address'],
'city': record['address']['city']
}
b.map(flatten).take(1)
Out[13]:
({'age': 29,
'occupation': 'Haulage Contractor',
'telephone': '113.919.6933',
'credit-card-number': '2471 7972 2251 8059',
'credit-card-expiration': '06/21',
'name': 'Preston Ray',
'street-address': '438 Southwood Spur',
'city': 'St. Charles'},)
In [14]:
df = b.map(flatten).to_dataframe()
df.head()
Out[14]:
| age | city | credit-card-expiration | credit-card-number | name | occupation | street-address | telephone | |
|---|---|---|---|---|---|---|---|---|
| 0 | 29 | St. Charles | 06/21 | 2471 7972 2251 8059 | Preston Ray | Haulage Contractor | 438 Southwood Spur | 113.919.6933 |
| 1 | 57 | Coronado | 09/25 | 3406 240769 42065 | Beverlee Price | Service Engineer | 326 San Bruno Stravenue | 045-216-5761 |
| 2 | 23 | Vallejo | 08/20 | 2530 0178 3032 2946 | Stacey Day | Aerobic Instructor | 266 Euclid Pike | 1-184-180-5009 |
| 3 | 65 | Rockwall | 10/17 | 2671 6820 1040 3714 | Shery Bradley | Bus Company | 253 Walkway Avenue | +1-(919)-146-4994 |
| 4 | 26 | Homestead | 03/17 | 4516 2827 7997 9973 | Shoshana Gould | Racehorse Groom | 1293 Mountain Spring Parkway | +1-(787)-615-3979 |
We can now perform the same computation as before, but now using Pandas and Dask dataframe.
In [15]:
df[df.age > 30].occupation.value_counts().nlargest(10).compute()
Out[15]:
Labelling Operator 14
Advertising Executive 14
Quantity Surveyor 14
Heating Engineer 14
Opera Singer 13
Landowner 13
Parts Supervisor 13
Hospital Worker 13
Barmaid 13
Screen Writer 13
Name: occupation, dtype: int64