Live Notebook
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.
[1]:
from dask.distributed import Client, progress
client = Client(n_workers=4, threads_per_worker=1)
client
/usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages/distributed/node.py:151: UserWarning: Port 8787 is already in use.
Perhaps you already have a cluster running?
Hosting the HTTP server on port 39077 instead
warnings.warn(
[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.
[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
[2]:
['/home/runner/work/dask-examples/dask-examples/data/0.json',
'/home/runner/work/dask-examples/dask-examples/data/1.json',
'/home/runner/work/dask-examples/dask-examples/data/2.json',
'/home/runner/work/dask-examples/dask-examples/data/3.json',
'/home/runner/work/dask-examples/dask-examples/data/4.json',
'/home/runner/work/dask-examples/dask-examples/data/5.json',
'/home/runner/work/dask-examples/dask-examples/data/6.json',
'/home/runner/work/dask-examples/dask-examples/data/7.json',
'/home/runner/work/dask-examples/dask-examples/data/8.json',
'/home/runner/work/dask-examples/dask-examples/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.
[3]:
!head -n 2 data/0.json
{"age": 45, "name": ["Ellyn", "Head"], "occupation": "Health Visitor", "telephone": "+1-(485)-639-9596", "address": {"address": "955 Flora Extension", "city": "Lake Zurich"}, "credit-card": {"number": "4193 2628 9870 3827", "expiration-date": "06/22"}}
{"age": 45, "name": ["Ta", "Schmidt"], "occupation": "Optical Advisor", "telephone": "357-161-9324", "address": {"address": "336 Sutro Heights Townline", "city": "Manchester"}, "credit-card": {"number": "3741 717460 63871", "expiration-date": "12/19"}}
[4]:
import dask.bag as db
import json
b = db.read_text('data/*.json').map(json.loads)
b
[4]:
dask.bag<loads, npartitions=10>
[5]:
b.take(2)
[5]:
({'age': 45,
'name': ['Ellyn', 'Head'],
'occupation': 'Health Visitor',
'telephone': '+1-(485)-639-9596',
'address': {'address': '955 Flora Extension', 'city': 'Lake Zurich'},
'credit-card': {'number': '4193 2628 9870 3827',
'expiration-date': '06/22'}},
{'age': 45,
'name': ['Ta', 'Schmidt'],
'occupation': 'Optical Advisor',
'telephone': '357-161-9324',
'address': {'address': '336 Sutro Heights Townline', 'city': 'Manchester'},
'credit-card': {'number': '3741 717460 63871', 'expiration-date': '12/19'}})
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.
[6]:
b.filter(lambda record: record['age'] > 30).take(2) # Select only people over 30
[6]:
({'age': 45,
'name': ['Ellyn', 'Head'],
'occupation': 'Health Visitor',
'telephone': '+1-(485)-639-9596',
'address': {'address': '955 Flora Extension', 'city': 'Lake Zurich'},
'credit-card': {'number': '4193 2628 9870 3827',
'expiration-date': '06/22'}},
{'age': 45,
'name': ['Ta', 'Schmidt'],
'occupation': 'Optical Advisor',
'telephone': '357-161-9324',
'address': {'address': '336 Sutro Heights Townline', 'city': 'Manchester'},
'credit-card': {'number': '3741 717460 63871', 'expiration-date': '12/19'}})
[7]:
b.map(lambda record: record['occupation']).take(2) # Select the occupation field
[7]:
('Health Visitor', 'Optical Advisor')
[8]:
b.count().compute() # Count total number of records
[8]:
10000
Chain computations¶
It is common to do many of these steps in one pipeline, only calling compute
or take
at the end.
[9]:
result = (b.filter(lambda record: record['age'] > 30)
.map(lambda record: record['occupation'])
.frequencies(sort=True)
.topk(10, key=1))
result
[9]:
dask.bag<topk-aggregate, 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.
[10]:
result.compute()
[10]:
[('Body Fitter', 15),
('Optical Advisor', 14),
('Research Consultant', 14),
('Telecommunication', 13),
('Haulage Contractor', 13),
('Gilder', 13),
('Quality Controller', 12),
('Induction Moulder', 12),
('Furniture Restorer', 12),
('Technical Manager', 12)]
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.
[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
[11]:
['/home/runner/work/dask-examples/dask-examples/data/processed.0.json',
'/home/runner/work/dask-examples/dask-examples/data/processed.1.json',
'/home/runner/work/dask-examples/dask-examples/data/processed.2.json',
'/home/runner/work/dask-examples/dask-examples/data/processed.3.json',
'/home/runner/work/dask-examples/dask-examples/data/processed.4.json',
'/home/runner/work/dask-examples/dask-examples/data/processed.5.json',
'/home/runner/work/dask-examples/dask-examples/data/processed.6.json',
'/home/runner/work/dask-examples/dask-examples/data/processed.7.json',
'/home/runner/work/dask-examples/dask-examples/data/processed.8.json',
'/home/runner/work/dask-examples/dask-examples/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.
[12]:
b.take(1)
[12]:
({'age': 45,
'name': ['Ellyn', 'Head'],
'occupation': 'Health Visitor',
'telephone': '+1-(485)-639-9596',
'address': {'address': '955 Flora Extension', 'city': 'Lake Zurich'},
'credit-card': {'number': '4193 2628 9870 3827',
'expiration-date': '06/22'}},)
[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)
[13]:
({'age': 45,
'occupation': 'Health Visitor',
'telephone': '+1-(485)-639-9596',
'credit-card-number': '4193 2628 9870 3827',
'credit-card-expiration': '06/22',
'name': 'Ellyn Head',
'street-address': '955 Flora Extension',
'city': 'Lake Zurich'},)
[14]:
df = b.map(flatten).to_dataframe()
df.head()
[14]:
age | occupation | telephone | credit-card-number | credit-card-expiration | name | street-address | city | |
---|---|---|---|---|---|---|---|---|
0 | 45 | Health Visitor | +1-(485)-639-9596 | 4193 2628 9870 3827 | 06/22 | Ellyn Head | 955 Flora Extension | Lake Zurich |
1 | 45 | Optical Advisor | 357-161-9324 | 3741 717460 63871 | 12/19 | Ta Schmidt | 336 Sutro Heights Townline | Manchester |
2 | 40 | Typist | +1-(524)-800-5717 | 2315 2650 3569 6447 | 11/24 | Jewell Newman | 83 Friendship Townline | Rock Springs |
3 | 66 | Technical Engineer | 975-610-6376 | 3798 841624 11839 | 12/22 | Kieth Clarke | 318 Marietta Private | Yukon |
4 | 27 | TV Editor | 854-587-3780 | 4342 2710 5891 2011 | 11/21 | Titus Holloway | 1175 State Mall | Richmond |
We can now perform the same computation as before, but now using Pandas and Dask dataframe.
[15]:
df[df.age > 30].occupation.value_counts().nlargest(10).compute()
[15]:
Body Fitter 15
Research Consultant 14
Optical Advisor 14
Gilder 13
Telecommunication 13
Haulage Contractor 13
Auto Electrician 12
Pharmacist 12
Ticket Agent 12
Quality Controller 12
Name: occupation, dtype: int64
Learn More¶
You may be interested in the following links:
dask tutorial, notebook 02, for a more in-depth introduction.