You can run this notebook in a live session or view it on Github.

Bag: Parallel Lists for semi-structured data

Dask-bag excels in processing data that can be represented as a sequence of arbitrary inputs. We’ll refer to this as “messy” data, because it can contain complex nested structures, missing fields, mixtures of data types, etc. The functional programming style fits very nicely with standard Python iteration, such as can be found in the itertools module.

Messy data is often encountered at the beginning of data processing pipelines when large volumes of raw data are first consumed. The initial set of data might be JSON, CSV, XML, or any other format that does not enforce strict structure and datatypes. For this reason, the initial data massaging and processing is often done with Python lists, dicts, and sets.

These core data structures are optimized for general-purpose storage and processing. Adding streaming computation with iterators/generator expressions or libraries like itertools or `toolz <https://toolz.readthedocs.io/en/latest/>`__ let us process large volumes in a small space. If we combine this with parallel processing then we can churn through a fair amount of data.

Dask.bag is a high level Dask collection to automate common workloads of this form. In a nutshell

dask.bag = map, filter, toolz + parallel execution

Related Documentation

• Bag documentation

• Bag screencast

• Bag API

• Bag examples

Create data

%run prep.py -d accounts

Setup

Again, we’ll use the distributed scheduler. Schedulers will be explained in depth later.

from dask.distributed import Client

client = Client(n_workers=4)

Creation

You can create a Bag from a Python sequence, from files, from data on S3, etc. We demonstrate using .take() to show elements of the data. (Doing .take(1) results in a tuple with one element)

Note that the data are partitioned into blocks, and there are many items per block. In the first example, the two partitions contain five elements each, and in the following two, each file is partitioned into one or more bytes blocks.

# each element is an integer
import dask.bag as db
b = db.from_sequence([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], npartitions=2)
b.take(3)

(1, 2, 3)

# each element is a text file, where each line is a JSON object
# note that the compression is handled automatically
import os
b = db.read_text(os.path.join('data', 'accounts.*.json.gz'))
b.take(1)

('{"id": 0, "name": "Alice", "transactions": [{"transaction-id": 17, "amount": 179}, {"transaction-id": 203, "amount": 113}, {"transaction-id": 302, "amount": -15}, {"transaction-id": 398, "amount": 204}, {"transaction-id": 987, "amount": 39}, {"transaction-id": 3434, "amount": 346}, {"transaction-id": 3568, "amount": 159}, {"transaction-id": 3597, "amount": 90}, {"transaction-id": 3817, "amount": 30}, {"transaction-id": 3824, "amount": 88}, {"transaction-id": 5497, "amount": 75}, {"transaction-id": 5512, "amount": 158}, {"transaction-id": 6095, "amount": 190}, {"transaction-id": 6192, "amount": 92}, {"transaction-id": 6752, "amount": 30}, {"transaction-id": 6778, "amount": 166}, {"transaction-id": 6940, "amount": -106}, {"transaction-id": 7057, "amount": 111}, {"transaction-id": 7151, "amount": 132}, {"transaction-id": 7863, "amount": 140}, {"transaction-id": 7917, "amount": 154}, {"transaction-id": 8291, "amount": 213}, {"transaction-id": 8294, "amount": 56}, {"transaction-id": 8652, "amount": 278}, {"transaction-id": 8767, "amount": 80}, {"transaction-id": 8955, "amount": 199}, {"transaction-id": 9124, "amount": 125}, {"transaction-id": 9368, "amount": 162}, {"transaction-id": 9834, "amount": 231}, {"transaction-id": 9849, "amount": 100}]}\n',)

# Edit sources.py to configure source locations
import sources
sources.bag_url

's3://dask-data/nyc-taxi/2015/yellow_tripdata_2015-01.csv'

# Requires `s3fs` library
# each partition is a remote CSV text file
b = db.read_text(sources.bag_url,
                 storage_options={'anon': True})
b.take(1)

('VendorID,tpep_pickup_datetime,tpep_dropoff_datetime,passenger_count,trip_distance,pickup_longitude,pickup_latitude,RateCodeID,store_and_fwd_flag,dropoff_longitude,dropoff_latitude,payment_type,fare_amount,extra,mta_tax,tip_amount,tolls_amount,improvement_surcharge,total_amount\n',)

Manipulation

Bag objects hold the standard functional API found in projects like the Python standard library, toolz, or pyspark, including map, filter, groupby, etc..

Operations on Bag objects create new bags. Call the .compute() method to trigger execution, as we saw for Delayed objects.

def is_even(n):
    return n % 2 == 0

b = db.from_sequence([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
c = b.filter(is_even).map(lambda x: x ** 2)
c

dask.bag<lambda, npartitions=10>

# blocking form: wait for completion (which is very fast in this case)
c.compute()

[4, 16, 36, 64, 100]

Example: Accounts JSON data

We’ve created a fake dataset of gzipped JSON data in your data directory. This is like the example used in the DataFrame example we will see later, except that it has bundled up all of the entries for each individual id into a single record. This is similar to data that you might collect off of a document store database or a web API.

Each line is a JSON encoded dictionary with the following keys

• id: Unique identifier of the customer

• name: Name of the customer

• transactions: List of transaction-id, amount pairs, one for each transaction for the customer in that file

filename = os.path.join('data', 'accounts.*.json.gz')
lines = db.read_text(filename)
lines.take(3)

('{"id": 0, "name": "Alice", "transactions": [{"transaction-id": 17, "amount": 179}, {"transaction-id": 203, "amount": 113}, {"transaction-id": 302, "amount": -15}, {"transaction-id": 398, "amount": 204}, {"transaction-id": 987, "amount": 39}, {"transaction-id": 3434, "amount": 346}, {"transaction-id": 3568, "amount": 159}, {"transaction-id": 3597, "amount": 90}, {"transaction-id": 3817, "amount": 30}, {"transaction-id": 3824, "amount": 88}, {"transaction-id": 5497, "amount": 75}, {"transaction-id": 5512, "amount": 158}, {"transaction-id": 6095, "amount": 190}, {"transaction-id": 6192, "amount": 92}, {"transaction-id": 6752, "amount": 30}, {"transaction-id": 6778, "amount": 166}, {"transaction-id": 6940, "amount": -106}, {"transaction-id": 7057, "amount": 111}, {"transaction-id": 7151, "amount": 132}, {"transaction-id": 7863, "amount": 140}, {"transaction-id": 7917, "amount": 154}, {"transaction-id": 8291, "amount": 213}, {"transaction-id": 8294, "amount": 56}, {"transaction-id": 8652, "amount": 278}, {"transaction-id": 8767, "amount": 80}, {"transaction-id": 8955, "amount": 199}, {"transaction-id": 9124, "amount": 125}, {"transaction-id": 9368, "amount": 162}, {"transaction-id": 9834, "amount": 231}, {"transaction-id": 9849, "amount": 100}]}\n',
 '{"id": 1, "name": "Frank", "transactions": [{"transaction-id": 143, "amount": 524}, {"transaction-id": 361, "amount": 477}, {"transaction-id": 649, "amount": 535}, {"transaction-id": 659, "amount": 496}, {"transaction-id": 1661, "amount": 548}, {"transaction-id": 2131, "amount": 411}, {"transaction-id": 2194, "amount": 390}, {"transaction-id": 2492, "amount": 513}, {"transaction-id": 2542, "amount": 405}, {"transaction-id": 3041, "amount": 517}, {"transaction-id": 3338, "amount": 483}, {"transaction-id": 5593, "amount": 466}, {"transaction-id": 5741, "amount": 417}, {"transaction-id": 5981, "amount": 522}, {"transaction-id": 6054, "amount": 416}, {"transaction-id": 6818, "amount": 543}, {"transaction-id": 8237, "amount": 336}, {"transaction-id": 9537, "amount": 470}, {"transaction-id": 9921, "amount": 527}]}\n',
 '{"id": 2, "name": "Bob", "transactions": [{"transaction-id": 76, "amount": 229}, {"transaction-id": 117, "amount": -155}, {"transaction-id": 306, "amount": 81}, {"transaction-id": 422, "amount": 0}, {"transaction-id": 448, "amount": 162}, {"transaction-id": 507, "amount": 213}, {"transaction-id": 545, "amount": -629}, {"transaction-id": 628, "amount": -377}, {"transaction-id": 655, "amount": 80}, {"transaction-id": 776, "amount": -1621}, {"transaction-id": 838, "amount": -517}, {"transaction-id": 906, "amount": 495}, {"transaction-id": 941, "amount": -993}, {"transaction-id": 997, "amount": 33}, {"transaction-id": 1072, "amount": -859}, {"transaction-id": 1105, "amount": -67}, {"transaction-id": 1326, "amount": -509}, {"transaction-id": 1672, "amount": -124}, {"transaction-id": 1697, "amount": -2}, {"transaction-id": 1730, "amount": 406}, {"transaction-id": 1871, "amount": 17}, {"transaction-id": 1879, "amount": -613}, {"transaction-id": 2012, "amount": 350}, {"transaction-id": 2041, "amount": -268}, {"transaction-id": 2074, "amount": -1194}, {"transaction-id": 2145, "amount": 33}, {"transaction-id": 2177, "amount": -81}, {"transaction-id": 2184, "amount": -848}, {"transaction-id": 2199, "amount": -448}, {"transaction-id": 2214, "amount": 84}, {"transaction-id": 2228, "amount": -272}, {"transaction-id": 2240, "amount": -381}, {"transaction-id": 2250, "amount": -808}, {"transaction-id": 2322, "amount": -495}, {"transaction-id": 2382, "amount": -1111}, {"transaction-id": 2384, "amount": -1131}, {"transaction-id": 2517, "amount": -74}, {"transaction-id": 2557, "amount": -476}, {"transaction-id": 2617, "amount": -727}, {"transaction-id": 2770, "amount": -853}, {"transaction-id": 2833, "amount": 136}, {"transaction-id": 2935, "amount": 121}, {"transaction-id": 3008, "amount": -56}, {"transaction-id": 3023, "amount": -674}, {"transaction-id": 3048, "amount": -834}, {"transaction-id": 3052, "amount": -362}, {"transaction-id": 3165, "amount": -911}, {"transaction-id": 3173, "amount": -420}, {"transaction-id": 3205, "amount": -271}, {"transaction-id": 3245, "amount": -1177}, {"transaction-id": 3261, "amount": 157}, {"transaction-id": 3309, "amount": 344}, {"transaction-id": 3370, "amount": -22}, {"transaction-id": 3461, "amount": -389}, {"transaction-id": 3608, "amount": -516}, {"transaction-id": 3648, "amount": 98}, {"transaction-id": 3680, "amount": 211}, {"transaction-id": 3770, "amount": -821}, {"transaction-id": 3785, "amount": 239}, {"transaction-id": 4190, "amount": 260}, {"transaction-id": 4221, "amount": 40}, {"transaction-id": 4229, "amount": -639}, {"transaction-id": 4313, "amount": 388}, {"transaction-id": 4381, "amount": -29}, {"transaction-id": 4472, "amount": -517}, {"transaction-id": 4480, "amount": -1}, {"transaction-id": 4593, "amount": -145}, {"transaction-id": 4602, "amount": -306}, {"transaction-id": 4603, "amount": -200}, {"transaction-id": 4661, "amount": -525}, {"transaction-id": 4676, "amount": -509}, {"transaction-id": 4857, "amount": -103}, {"transaction-id": 4940, "amount": -1165}, {"transaction-id": 4959, "amount": -774}, {"transaction-id": 5092, "amount": -641}, {"transaction-id": 5193, "amount": 440}, {"transaction-id": 5254, "amount": -657}, {"transaction-id": 5423, "amount": -607}, {"transaction-id": 5500, "amount": -522}, {"transaction-id": 5573, "amount": -186}, {"transaction-id": 5665, "amount": -240}, {"transaction-id": 5680, "amount": -1624}, {"transaction-id": 5682, "amount": 550}, {"transaction-id": 5899, "amount": -227}, {"transaction-id": 5983, "amount": -660}, {"transaction-id": 5986, "amount": -389}, {"transaction-id": 5999, "amount": -912}, {"transaction-id": 6107, "amount": -648}, {"transaction-id": 6178, "amount": -554}, {"transaction-id": 6218, "amount": -604}, {"transaction-id": 6381, "amount": -1284}, {"transaction-id": 6511, "amount": 485}, {"transaction-id": 6524, "amount": -802}, {"transaction-id": 6810, "amount": -527}, {"transaction-id": 6929, "amount": 91}, {"transaction-id": 7039, "amount": -672}, {"transaction-id": 7263, "amount": -1097}, {"transaction-id": 7289, "amount": -692}, {"transaction-id": 7298, "amount": 108}, {"transaction-id": 7365, "amount": 86}, {"transaction-id": 7397, "amount": -588}, {"transaction-id": 7442, "amount": -281}, {"transaction-id": 7599, "amount": 170}, {"transaction-id": 7668, "amount": -505}, {"transaction-id": 7790, "amount": -1011}, {"transaction-id": 7812, "amount": -801}, {"transaction-id": 7867, "amount": -260}, {"transaction-id": 7877, "amount": -578}, {"transaction-id": 7915, "amount": -580}, {"transaction-id": 8119, "amount": -580}, {"transaction-id": 8123, "amount": -250}, {"transaction-id": 8164, "amount": -684}, {"transaction-id": 8234, "amount": -769}, {"transaction-id": 8243, "amount": -562}, {"transaction-id": 8275, "amount": 144}, {"transaction-id": 8299, "amount": -845}, {"transaction-id": 8422, "amount": 343}, {"transaction-id": 8461, "amount": -399}, {"transaction-id": 8510, "amount": -501}, {"transaction-id": 8598, "amount": -288}, {"transaction-id": 8814, "amount": 339}, {"transaction-id": 8838, "amount": -746}, {"transaction-id": 8880, "amount": -412}, {"transaction-id": 8969, "amount": -297}, {"transaction-id": 8994, "amount": -279}, {"transaction-id": 9009, "amount": -275}, {"transaction-id": 9148, "amount": 34}, {"transaction-id": 9212, "amount": -205}, {"transaction-id": 9230, "amount": -695}, {"transaction-id": 9442, "amount": 305}, {"transaction-id": 9534, "amount": -741}, {"transaction-id": 9918, "amount": -1110}, {"transaction-id": 9957, "amount": -431}]}\n')

Our data comes out of the file as lines of text. Notice that file decompression happened automatically. We can make this data look more reasonable by mapping the json.loads function onto our bag.

import json
js = lines.map(json.loads)
# take: inspect first few elements
js.take(3)

({'id': 0,
  'name': 'Alice',
  'transactions': [{'transaction-id': 17, 'amount': 179},
   {'transaction-id': 203, 'amount': 113},
   {'transaction-id': 302, 'amount': -15},
   {'transaction-id': 398, 'amount': 204},
   {'transaction-id': 987, 'amount': 39},
   {'transaction-id': 3434, 'amount': 346},
   {'transaction-id': 3568, 'amount': 159},
   {'transaction-id': 3597, 'amount': 90},
   {'transaction-id': 3817, 'amount': 30},
   {'transaction-id': 3824, 'amount': 88},
   {'transaction-id': 5497, 'amount': 75},
   {'transaction-id': 5512, 'amount': 158},
   {'transaction-id': 6095, 'amount': 190},
   {'transaction-id': 6192, 'amount': 92},
   {'transaction-id': 6752, 'amount': 30},
   {'transaction-id': 6778, 'amount': 166},
   {'transaction-id': 6940, 'amount': -106},
   {'transaction-id': 7057, 'amount': 111},
   {'transaction-id': 7151, 'amount': 132},
   {'transaction-id': 7863, 'amount': 140},
   {'transaction-id': 7917, 'amount': 154},
   {'transaction-id': 8291, 'amount': 213},
   {'transaction-id': 8294, 'amount': 56},
   {'transaction-id': 8652, 'amount': 278},
   {'transaction-id': 8767, 'amount': 80},
   {'transaction-id': 8955, 'amount': 199},
   {'transaction-id': 9124, 'amount': 125},
   {'transaction-id': 9368, 'amount': 162},
   {'transaction-id': 9834, 'amount': 231},
   {'transaction-id': 9849, 'amount': 100}]},
 {'id': 1,
  'name': 'Frank',
  'transactions': [{'transaction-id': 143, 'amount': 524},
   {'transaction-id': 361, 'amount': 477},
   {'transaction-id': 649, 'amount': 535},
   {'transaction-id': 659, 'amount': 496},
   {'transaction-id': 1661, 'amount': 548},
   {'transaction-id': 2131, 'amount': 411},
   {'transaction-id': 2194, 'amount': 390},
   {'transaction-id': 2492, 'amount': 513},
   {'transaction-id': 2542, 'amount': 405},
   {'transaction-id': 3041, 'amount': 517},
   {'transaction-id': 3338, 'amount': 483},
   {'transaction-id': 5593, 'amount': 466},
   {'transaction-id': 5741, 'amount': 417},
   {'transaction-id': 5981, 'amount': 522},
   {'transaction-id': 6054, 'amount': 416},
   {'transaction-id': 6818, 'amount': 543},
   {'transaction-id': 8237, 'amount': 336},
   {'transaction-id': 9537, 'amount': 470},
   {'transaction-id': 9921, 'amount': 527}]},
 {'id': 2,
  'name': 'Bob',
  'transactions': [{'transaction-id': 76, 'amount': 229},
   {'transaction-id': 117, 'amount': -155},
   {'transaction-id': 306, 'amount': 81},
   {'transaction-id': 422, 'amount': 0},
   {'transaction-id': 448, 'amount': 162},
   {'transaction-id': 507, 'amount': 213},
   {'transaction-id': 545, 'amount': -629},
   {'transaction-id': 628, 'amount': -377},
   {'transaction-id': 655, 'amount': 80},
   {'transaction-id': 776, 'amount': -1621},
   {'transaction-id': 838, 'amount': -517},
   {'transaction-id': 906, 'amount': 495},
   {'transaction-id': 941, 'amount': -993},
   {'transaction-id': 997, 'amount': 33},
   {'transaction-id': 1072, 'amount': -859},
   {'transaction-id': 1105, 'amount': -67},
   {'transaction-id': 1326, 'amount': -509},
   {'transaction-id': 1672, 'amount': -124},
   {'transaction-id': 1697, 'amount': -2},
   {'transaction-id': 1730, 'amount': 406},
   {'transaction-id': 1871, 'amount': 17},
   {'transaction-id': 1879, 'amount': -613},
   {'transaction-id': 2012, 'amount': 350},
   {'transaction-id': 2041, 'amount': -268},
   {'transaction-id': 2074, 'amount': -1194},
   {'transaction-id': 2145, 'amount': 33},
   {'transaction-id': 2177, 'amount': -81},
   {'transaction-id': 2184, 'amount': -848},
   {'transaction-id': 2199, 'amount': -448},
   {'transaction-id': 2214, 'amount': 84},
   {'transaction-id': 2228, 'amount': -272},
   {'transaction-id': 2240, 'amount': -381},
   {'transaction-id': 2250, 'amount': -808},
   {'transaction-id': 2322, 'amount': -495},
   {'transaction-id': 2382, 'amount': -1111},
   {'transaction-id': 2384, 'amount': -1131},
   {'transaction-id': 2517, 'amount': -74},
   {'transaction-id': 2557, 'amount': -476},
   {'transaction-id': 2617, 'amount': -727},
   {'transaction-id': 2770, 'amount': -853},
   {'transaction-id': 2833, 'amount': 136},
   {'transaction-id': 2935, 'amount': 121},
   {'transaction-id': 3008, 'amount': -56},
   {'transaction-id': 3023, 'amount': -674},
   {'transaction-id': 3048, 'amount': -834},
   {'transaction-id': 3052, 'amount': -362},
   {'transaction-id': 3165, 'amount': -911},
   {'transaction-id': 3173, 'amount': -420},
   {'transaction-id': 3205, 'amount': -271},
   {'transaction-id': 3245, 'amount': -1177},
   {'transaction-id': 3261, 'amount': 157},
   {'transaction-id': 3309, 'amount': 344},
   {'transaction-id': 3370, 'amount': -22},
   {'transaction-id': 3461, 'amount': -389},
   {'transaction-id': 3608, 'amount': -516},
   {'transaction-id': 3648, 'amount': 98},
   {'transaction-id': 3680, 'amount': 211},
   {'transaction-id': 3770, 'amount': -821},
   {'transaction-id': 3785, 'amount': 239},
   {'transaction-id': 4190, 'amount': 260},
   {'transaction-id': 4221, 'amount': 40},
   {'transaction-id': 4229, 'amount': -639},
   {'transaction-id': 4313, 'amount': 388},
   {'transaction-id': 4381, 'amount': -29},
   {'transaction-id': 4472, 'amount': -517},
   {'transaction-id': 4480, 'amount': -1},
   {'transaction-id': 4593, 'amount': -145},
   {'transaction-id': 4602, 'amount': -306},
   {'transaction-id': 4603, 'amount': -200},
   {'transaction-id': 4661, 'amount': -525},
   {'transaction-id': 4676, 'amount': -509},
   {'transaction-id': 4857, 'amount': -103},
   {'transaction-id': 4940, 'amount': -1165},
   {'transaction-id': 4959, 'amount': -774},
   {'transaction-id': 5092, 'amount': -641},
   {'transaction-id': 5193, 'amount': 440},
   {'transaction-id': 5254, 'amount': -657},
   {'transaction-id': 5423, 'amount': -607},
   {'transaction-id': 5500, 'amount': -522},
   {'transaction-id': 5573, 'amount': -186},
   {'transaction-id': 5665, 'amount': -240},
   {'transaction-id': 5680, 'amount': -1624},
   {'transaction-id': 5682, 'amount': 550},
   {'transaction-id': 5899, 'amount': -227},
   {'transaction-id': 5983, 'amount': -660},
   {'transaction-id': 5986, 'amount': -389},
   {'transaction-id': 5999, 'amount': -912},
   {'transaction-id': 6107, 'amount': -648},
   {'transaction-id': 6178, 'amount': -554},
   {'transaction-id': 6218, 'amount': -604},
   {'transaction-id': 6381, 'amount': -1284},
   {'transaction-id': 6511, 'amount': 485},
   {'transaction-id': 6524, 'amount': -802},
   {'transaction-id': 6810, 'amount': -527},
   {'transaction-id': 6929, 'amount': 91},
   {'transaction-id': 7039, 'amount': -672},
   {'transaction-id': 7263, 'amount': -1097},
   {'transaction-id': 7289, 'amount': -692},
   {'transaction-id': 7298, 'amount': 108},
   {'transaction-id': 7365, 'amount': 86},
   {'transaction-id': 7397, 'amount': -588},
   {'transaction-id': 7442, 'amount': -281},
   {'transaction-id': 7599, 'amount': 170},
   {'transaction-id': 7668, 'amount': -505},
   {'transaction-id': 7790, 'amount': -1011},
   {'transaction-id': 7812, 'amount': -801},
   {'transaction-id': 7867, 'amount': -260},
   {'transaction-id': 7877, 'amount': -578},
   {'transaction-id': 7915, 'amount': -580},
   {'transaction-id': 8119, 'amount': -580},
   {'transaction-id': 8123, 'amount': -250},
   {'transaction-id': 8164, 'amount': -684},
   {'transaction-id': 8234, 'amount': -769},
   {'transaction-id': 8243, 'amount': -562},
   {'transaction-id': 8275, 'amount': 144},
   {'transaction-id': 8299, 'amount': -845},
   {'transaction-id': 8422, 'amount': 343},
   {'transaction-id': 8461, 'amount': -399},
   {'transaction-id': 8510, 'amount': -501},
   {'transaction-id': 8598, 'amount': -288},
   {'transaction-id': 8814, 'amount': 339},
   {'transaction-id': 8838, 'amount': -746},
   {'transaction-id': 8880, 'amount': -412},
   {'transaction-id': 8969, 'amount': -297},
   {'transaction-id': 8994, 'amount': -279},
   {'transaction-id': 9009, 'amount': -275},
   {'transaction-id': 9148, 'amount': 34},
   {'transaction-id': 9212, 'amount': -205},
   {'transaction-id': 9230, 'amount': -695},
   {'transaction-id': 9442, 'amount': 305},
   {'transaction-id': 9534, 'amount': -741},
   {'transaction-id': 9918, 'amount': -1110},
   {'transaction-id': 9957, 'amount': -431}]})

Basic Queries

Once we parse our JSON data into proper Python objects (dicts, lists, etc.) we can perform more interesting queries by creating small Python functions to run on our data.

# filter: keep only some elements of the sequence
js.filter(lambda record: record['name'] == 'Alice').take(5)

({'id': 0,
  'name': 'Alice',
  'transactions': [{'transaction-id': 17, 'amount': 179},
   {'transaction-id': 203, 'amount': 113},
   {'transaction-id': 302, 'amount': -15},
   {'transaction-id': 398, 'amount': 204},
   {'transaction-id': 987, 'amount': 39},
   {'transaction-id': 3434, 'amount': 346},
   {'transaction-id': 3568, 'amount': 159},
   {'transaction-id': 3597, 'amount': 90},
   {'transaction-id': 3817, 'amount': 30},
   {'transaction-id': 3824, 'amount': 88},
   {'transaction-id': 5497, 'amount': 75},
   {'transaction-id': 5512, 'amount': 158},
   {'transaction-id': 6095, 'amount': 190},
   {'transaction-id': 6192, 'amount': 92},
   {'transaction-id': 6752, 'amount': 30},
   {'transaction-id': 6778, 'amount': 166},
   {'transaction-id': 6940, 'amount': -106},
   {'transaction-id': 7057, 'amount': 111},
   {'transaction-id': 7151, 'amount': 132},
   {'transaction-id': 7863, 'amount': 140},
   {'transaction-id': 7917, 'amount': 154},
   {'transaction-id': 8291, 'amount': 213},
   {'transaction-id': 8294, 'amount': 56},
   {'transaction-id': 8652, 'amount': 278},
   {'transaction-id': 8767, 'amount': 80},
   {'transaction-id': 8955, 'amount': 199},
   {'transaction-id': 9124, 'amount': 125},
   {'transaction-id': 9368, 'amount': 162},
   {'transaction-id': 9834, 'amount': 231},
   {'transaction-id': 9849, 'amount': 100}]},
 {'id': 21,
  'name': 'Alice',
  'transactions': [{'transaction-id': 66, 'amount': 490},
   {'transaction-id': 107, 'amount': 465},
   {'transaction-id': 109, 'amount': 494},
   {'transaction-id': 277, 'amount': 483},
   {'transaction-id': 835, 'amount': 548},
   {'transaction-id': 876, 'amount': 452},
   {'transaction-id': 1205, 'amount': 469},
   {'transaction-id': 1216, 'amount': 454},
   {'transaction-id': 1266, 'amount': 496},
   {'transaction-id': 1521, 'amount': 394},
   {'transaction-id': 1702, 'amount': 533},
   {'transaction-id': 1863, 'amount': 452},
   {'transaction-id': 2043, 'amount': 449},
   {'transaction-id': 2274, 'amount': 451},
   {'transaction-id': 3026, 'amount': 491},
   {'transaction-id': 3690, 'amount': 464},
   {'transaction-id': 3831, 'amount': 494},
   {'transaction-id': 3888, 'amount': 508},
   {'transaction-id': 4098, 'amount': 508},
   {'transaction-id': 4513, 'amount': 505},
   {'transaction-id': 4528, 'amount': 441},
   {'transaction-id': 4658, 'amount': 492},
   {'transaction-id': 4785, 'amount': 434},
   {'transaction-id': 4978, 'amount': 492},
   {'transaction-id': 5018, 'amount': 478},
   {'transaction-id': 5047, 'amount': 415},
   {'transaction-id': 5090, 'amount': 506},
   {'transaction-id': 5224, 'amount': 468},
   {'transaction-id': 5352, 'amount': 475},
   {'transaction-id': 5471, 'amount': 459},
   {'transaction-id': 5582, 'amount': 431},
   {'transaction-id': 5590, 'amount': 537},
   {'transaction-id': 5645, 'amount': 446},
   {'transaction-id': 5703, 'amount': 461},
   {'transaction-id': 5911, 'amount': 466},
   {'transaction-id': 6071, 'amount': 417},
   {'transaction-id': 6255, 'amount': 520},
   {'transaction-id': 6267, 'amount': 571},
   {'transaction-id': 6756, 'amount': 410},
   {'transaction-id': 6865, 'amount': 528},
   {'transaction-id': 6982, 'amount': 491},
   {'transaction-id': 7021, 'amount': 461},
   {'transaction-id': 7108, 'amount': 514},
   {'transaction-id': 7247, 'amount': 450},
   {'transaction-id': 7564, 'amount': 435},
   {'transaction-id': 7610, 'amount': 480},
   {'transaction-id': 7681, 'amount': 496},
   {'transaction-id': 7736, 'amount': 492},
   {'transaction-id': 7912, 'amount': 424},
   {'transaction-id': 8129, 'amount': 444},
   {'transaction-id': 8158, 'amount': 488},
   {'transaction-id': 8195, 'amount': 467},
   {'transaction-id': 8253, 'amount': 388},
   {'transaction-id': 8380, 'amount': 478},
   {'transaction-id': 8386, 'amount': 409},
   {'transaction-id': 8569, 'amount': 473},
   {'transaction-id': 8721, 'amount': 487},
   {'transaction-id': 8788, 'amount': 414},
   {'transaction-id': 8979, 'amount': 503},
   {'transaction-id': 9043, 'amount': 426},
   {'transaction-id': 9073, 'amount': 472},
   {'transaction-id': 9594, 'amount': 446},
   {'transaction-id': 9763, 'amount': 494},
   {'transaction-id': 9774, 'amount': 479}]},
 {'id': 30,
  'name': 'Alice',
  'transactions': [{'transaction-id': 1798, 'amount': 690},
   {'transaction-id': 2168, 'amount': 748},
   {'transaction-id': 2216, 'amount': 760},
   {'transaction-id': 2301, 'amount': 767},
   {'transaction-id': 2825, 'amount': 721},
   {'transaction-id': 4485, 'amount': 648},
   {'transaction-id': 5335, 'amount': 760},
   {'transaction-id': 5446, 'amount': 710},
   {'transaction-id': 5515, 'amount': 681},
   {'transaction-id': 6261, 'amount': 699},
   {'transaction-id': 6388, 'amount': 748},
   {'transaction-id': 6538, 'amount': 732},
   {'transaction-id': 6963, 'amount': 718},
   {'transaction-id': 7532, 'amount': 810},
   {'transaction-id': 7983, 'amount': 782},
   {'transaction-id': 8372, 'amount': 728},
   {'transaction-id': 8602, 'amount': 757},
   {'transaction-id': 8685, 'amount': 732},
   {'transaction-id': 8877, 'amount': 726},
   {'transaction-id': 8961, 'amount': 814},
   {'transaction-id': 8981, 'amount': 748},
   {'transaction-id': 9403, 'amount': 752},
   {'transaction-id': 9644, 'amount': 773},
   {'transaction-id': 9886, 'amount': 707}]},
 {'id': 41,
  'name': 'Alice',
  'transactions': [{'transaction-id': 77, 'amount': 528},
   {'transaction-id': 259, 'amount': 577},
   {'transaction-id': 802, 'amount': 522},
   {'transaction-id': 848, 'amount': 554},
   {'transaction-id': 1163, 'amount': 536},
   {'transaction-id': 1533, 'amount': 525},
   {'transaction-id': 1553, 'amount': 522},
   {'transaction-id': 1899, 'amount': 568},
   {'transaction-id': 2195, 'amount': 536},
   {'transaction-id': 2238, 'amount': 522},
   {'transaction-id': 2288, 'amount': 579},
   {'transaction-id': 2532, 'amount': 522},
   {'transaction-id': 2983, 'amount': 601},
   {'transaction-id': 4053, 'amount': 538},
   {'transaction-id': 4100, 'amount': 553},
   {'transaction-id': 4208, 'amount': 544},
   {'transaction-id': 4385, 'amount': 574},
   {'transaction-id': 4589, 'amount': 522},
   {'transaction-id': 4646, 'amount': 523},
   {'transaction-id': 4737, 'amount': 553},
   {'transaction-id': 5035, 'amount': 591},
   {'transaction-id': 5747, 'amount': 558},
   {'transaction-id': 6313, 'amount': 565},
   {'transaction-id': 6686, 'amount': 571},
   {'transaction-id': 6918, 'amount': 531},
   {'transaction-id': 7006, 'amount': 542},
   {'transaction-id': 7410, 'amount': 549},
   {'transaction-id': 7417, 'amount': 523},
   {'transaction-id': 7770, 'amount': 522},
   {'transaction-id': 7996, 'amount': 534},
   {'transaction-id': 8044, 'amount': 529},
   {'transaction-id': 8186, 'amount': 546},
   {'transaction-id': 8445, 'amount': 539},
   {'transaction-id': 8612, 'amount': 527},
   {'transaction-id': 8775, 'amount': 552},
   {'transaction-id': 8874, 'amount': 512}]},
 {'id': 49,
  'name': 'Alice',
  'transactions': [{'transaction-id': 650, 'amount': 351},
   {'transaction-id': 1192, 'amount': 460},
   {'transaction-id': 1249, 'amount': 355},
   {'transaction-id': 2097, 'amount': 458},
   {'transaction-id': 2754, 'amount': 381},
   {'transaction-id': 3328, 'amount': 390},
   {'transaction-id': 3662, 'amount': 379},
   {'transaction-id': 4099, 'amount': 358},
   {'transaction-id': 4744, 'amount': 325},
   {'transaction-id': 4860, 'amount': 352},
   {'transaction-id': 5460, 'amount': 326},
   {'transaction-id': 5479, 'amount': 283},
   {'transaction-id': 5704, 'amount': 225},
   {'transaction-id': 7048, 'amount': 347},
   {'transaction-id': 7487, 'amount': 433},
   {'transaction-id': 7653, 'amount': 284},
   {'transaction-id': 8454, 'amount': 274},
   {'transaction-id': 8633, 'amount': 306},
   {'transaction-id': 9300, 'amount': 278}]})

def count_transactions(d):
    return {'name': d['name'], 'count': len(d['transactions'])}

# map: apply a function to each element
(js.filter(lambda record: record['name'] == 'Alice')
   .map(count_transactions)
   .take(5))

({'name': 'Alice', 'count': 30},
 {'name': 'Alice', 'count': 64},
 {'name': 'Alice', 'count': 24},
 {'name': 'Alice', 'count': 36},
 {'name': 'Alice', 'count': 19})

# pluck: select a field, as from a dictionary, element[field]
(js.filter(lambda record: record['name'] == 'Alice')
   .map(count_transactions)
   .pluck('count')
   .take(5))

(30, 64, 24, 36, 19)

# Average number of transactions for all of the Alice entries
(js.filter(lambda record: record['name'] == 'Alice')
   .map(count_transactions)
   .pluck('count')
   .mean()
   .compute())

37.58714285714286

Use flatten to de-nest

In the example below we see the use of .flatten() to flatten results. We compute the average amount for all transactions for all Alices.

(js.filter(lambda record: record['name'] == 'Alice')
   .pluck('transactions')
   .take(3))

([{'transaction-id': 17, 'amount': 179},
  {'transaction-id': 203, 'amount': 113},
  {'transaction-id': 302, 'amount': -15},
  {'transaction-id': 398, 'amount': 204},
  {'transaction-id': 987, 'amount': 39},
  {'transaction-id': 3434, 'amount': 346},
  {'transaction-id': 3568, 'amount': 159},
  {'transaction-id': 3597, 'amount': 90},
  {'transaction-id': 3817, 'amount': 30},
  {'transaction-id': 3824, 'amount': 88},
  {'transaction-id': 5497, 'amount': 75},
  {'transaction-id': 5512, 'amount': 158},
  {'transaction-id': 6095, 'amount': 190},
  {'transaction-id': 6192, 'amount': 92},
  {'transaction-id': 6752, 'amount': 30},
  {'transaction-id': 6778, 'amount': 166},
  {'transaction-id': 6940, 'amount': -106},
  {'transaction-id': 7057, 'amount': 111},
  {'transaction-id': 7151, 'amount': 132},
  {'transaction-id': 7863, 'amount': 140},
  {'transaction-id': 7917, 'amount': 154},
  {'transaction-id': 8291, 'amount': 213},
  {'transaction-id': 8294, 'amount': 56},
  {'transaction-id': 8652, 'amount': 278},
  {'transaction-id': 8767, 'amount': 80},
  {'transaction-id': 8955, 'amount': 199},
  {'transaction-id': 9124, 'amount': 125},
  {'transaction-id': 9368, 'amount': 162},
  {'transaction-id': 9834, 'amount': 231},
  {'transaction-id': 9849, 'amount': 100}],
 [{'transaction-id': 66, 'amount': 490},
  {'transaction-id': 107, 'amount': 465},
  {'transaction-id': 109, 'amount': 494},
  {'transaction-id': 277, 'amount': 483},
  {'transaction-id': 835, 'amount': 548},
  {'transaction-id': 876, 'amount': 452},
  {'transaction-id': 1205, 'amount': 469},
  {'transaction-id': 1216, 'amount': 454},
  {'transaction-id': 1266, 'amount': 496},
  {'transaction-id': 1521, 'amount': 394},
  {'transaction-id': 1702, 'amount': 533},
  {'transaction-id': 1863, 'amount': 452},
  {'transaction-id': 2043, 'amount': 449},
  {'transaction-id': 2274, 'amount': 451},
  {'transaction-id': 3026, 'amount': 491},
  {'transaction-id': 3690, 'amount': 464},
  {'transaction-id': 3831, 'amount': 494},
  {'transaction-id': 3888, 'amount': 508},
  {'transaction-id': 4098, 'amount': 508},
  {'transaction-id': 4513, 'amount': 505},
  {'transaction-id': 4528, 'amount': 441},
  {'transaction-id': 4658, 'amount': 492},
  {'transaction-id': 4785, 'amount': 434},
  {'transaction-id': 4978, 'amount': 492},
  {'transaction-id': 5018, 'amount': 478},
  {'transaction-id': 5047, 'amount': 415},
  {'transaction-id': 5090, 'amount': 506},
  {'transaction-id': 5224, 'amount': 468},
  {'transaction-id': 5352, 'amount': 475},
  {'transaction-id': 5471, 'amount': 459},
  {'transaction-id': 5582, 'amount': 431},
  {'transaction-id': 5590, 'amount': 537},
  {'transaction-id': 5645, 'amount': 446},
  {'transaction-id': 5703, 'amount': 461},
  {'transaction-id': 5911, 'amount': 466},
  {'transaction-id': 6071, 'amount': 417},
  {'transaction-id': 6255, 'amount': 520},
  {'transaction-id': 6267, 'amount': 571},
  {'transaction-id': 6756, 'amount': 410},
  {'transaction-id': 6865, 'amount': 528},
  {'transaction-id': 6982, 'amount': 491},
  {'transaction-id': 7021, 'amount': 461},
  {'transaction-id': 7108, 'amount': 514},
  {'transaction-id': 7247, 'amount': 450},
  {'transaction-id': 7564, 'amount': 435},
  {'transaction-id': 7610, 'amount': 480},
  {'transaction-id': 7681, 'amount': 496},
  {'transaction-id': 7736, 'amount': 492},
  {'transaction-id': 7912, 'amount': 424},
  {'transaction-id': 8129, 'amount': 444},
  {'transaction-id': 8158, 'amount': 488},
  {'transaction-id': 8195, 'amount': 467},
  {'transaction-id': 8253, 'amount': 388},
  {'transaction-id': 8380, 'amount': 478},
  {'transaction-id': 8386, 'amount': 409},
  {'transaction-id': 8569, 'amount': 473},
  {'transaction-id': 8721, 'amount': 487},
  {'transaction-id': 8788, 'amount': 414},
  {'transaction-id': 8979, 'amount': 503},
  {'transaction-id': 9043, 'amount': 426},
  {'transaction-id': 9073, 'amount': 472},
  {'transaction-id': 9594, 'amount': 446},
  {'transaction-id': 9763, 'amount': 494},
  {'transaction-id': 9774, 'amount': 479}],
 [{'transaction-id': 1798, 'amount': 690},
  {'transaction-id': 2168, 'amount': 748},
  {'transaction-id': 2216, 'amount': 760},
  {'transaction-id': 2301, 'amount': 767},
  {'transaction-id': 2825, 'amount': 721},
  {'transaction-id': 4485, 'amount': 648},
  {'transaction-id': 5335, 'amount': 760},
  {'transaction-id': 5446, 'amount': 710},
  {'transaction-id': 5515, 'amount': 681},
  {'transaction-id': 6261, 'amount': 699},
  {'transaction-id': 6388, 'amount': 748},
  {'transaction-id': 6538, 'amount': 732},
  {'transaction-id': 6963, 'amount': 718},
  {'transaction-id': 7532, 'amount': 810},
  {'transaction-id': 7983, 'amount': 782},
  {'transaction-id': 8372, 'amount': 728},
  {'transaction-id': 8602, 'amount': 757},
  {'transaction-id': 8685, 'amount': 732},
  {'transaction-id': 8877, 'amount': 726},
  {'transaction-id': 8961, 'amount': 814},
  {'transaction-id': 8981, 'amount': 748},
  {'transaction-id': 9403, 'amount': 752},
  {'transaction-id': 9644, 'amount': 773},
  {'transaction-id': 9886, 'amount': 707}])

(js.filter(lambda record: record['name'] == 'Alice')
   .pluck('transactions')
   .flatten()
   .take(3))

({'transaction-id': 17, 'amount': 179},
 {'transaction-id': 203, 'amount': 113},
 {'transaction-id': 302, 'amount': -15})

(js.filter(lambda record: record['name'] == 'Alice')
   .pluck('transactions')
   .flatten()
   .pluck('amount')
   .take(3))

(179, 113, -15)

(js.filter(lambda record: record['name'] == 'Alice')
   .pluck('transactions')
   .flatten()
   .pluck('amount')
   .mean()
   .compute())

1415.7387784576792

Groupby and Foldby

Often we want to group data by some function or key. We can do this either with the .groupby method, which is straightforward but forces a full shuffle of the data (expensive) or with the harder-to-use but faster .foldby method, which does a streaming combined groupby and reduction.

• groupby: Shuffles data so that all items with the same key are in the same key-value pair

• foldby: Walks through the data accumulating a result per key

Note: the full groupby is particularly bad. In actual workloads you would do well to use ``foldby`` or switch to ``DataFrame``s if possible.

groupby

Groupby collects items in your collection so that all items with the same value under some function are collected together into a key-value pair.

b = db.from_sequence(['Alice', 'Bob', 'Charlie', 'Dan', 'Edith', 'Frank'])
b.groupby(len).compute()  # names grouped by length

[(7, ['Charlie']), (3, ['Bob', 'Dan']), (5, ['Alice', 'Edith', 'Frank'])]

b = db.from_sequence(list(range(10)))
b.groupby(lambda x: x % 2).compute()

[(0, [0, 2, 4, 6, 8]), (1, [1, 3, 5, 7, 9])]

b.groupby(lambda x: x % 2).starmap(lambda k, v: (k, max(v))).compute()

[(0, 8), (1, 9)]

foldby

Foldby can be quite odd at first. It is similar to the following functions from other libraries:

• `toolz.reduceby <http://toolz.readthedocs.io/en/latest/streaming-analytics.html#streaming-split-apply-combine>`__

• `pyspark.RDD.combineByKey <http://abshinn.github.io/python/apache-spark/2014/10/11/using-combinebykey-in-apache-spark/>`__

When using foldby you provide

1. A key function on which to group elements

2. A binary operator such as you would pass to reduce that you use to perform reduction per each group

3. A combine binary operator that can combine the results of two reduce calls on different parts of your dataset.

Your reduction must be associative. It will happen in parallel in each of the partitions of your dataset. Then all of these intermediate results will be combined by the combine binary operator.

b.foldby(lambda x: x % 2, binop=max, combine=max).compute()

[(0, 8), (1, 9)]

Example with account data

We find the number of people with the same name.

%%time
# Warning, this one takes a while...
result = js.groupby(lambda item: item['name']).starmap(lambda k, v: (k, len(v))).compute()
print(sorted(result))

[('Alice', 154), ('Alice', 168), ('Alice', 182), ('Alice', 196), ('Bob', 86), ('Bob', 94), ('Bob', 99), ('Bob', 107), ('Charlie', 99), ('Charlie', 108), ('Charlie', 117), ('Charlie', 126), ('Dan', 81), ('Dan', 86), ('Dan', 170), ('Edith', 110), ('Edith', 120), ('Edith', 130), ('Edith', 140), ('Frank', 33), ('Frank', 36), ('Frank', 39), ('Frank', 42), ('George', 121), ('George', 132), ('George', 143), ('George', 154), ('Hannah', 70), ('Hannah', 78), ('Hannah', 81), ('Hannah', 90), ('Ingrid', 55), ('Ingrid', 60), ('Ingrid', 65), ('Ingrid', 70), ('Jerry', 180), ('Jerry', 210), ('Jerry', 360), ('Kevin', 42), ('Kevin', 47), ('Kevin', 51), ('Kevin', 54), ('Laura', 165), ('Laura', 180), ('Laura', 195), ('Laura', 210), ('Michael', 150), ('Michael', 166), ('Michael', 181), ('Michael', 194), ('Norbert', 66), ('Norbert', 72), ('Norbert', 78), ('Norbert', 84), ('Oliver', 110), ('Oliver', 118), ('Oliver', 128), ('Oliver', 138), ('Patricia', 116), ('Patricia', 124), ('Patricia', 134), ('Patricia', 147), ('Quinn', 146), ('Quinn', 162), ('Quinn', 176), ('Quinn', 187), ('Ray', 121), ('Ray', 131), ('Ray', 143), ('Ray', 153), ('Sarah', 77), ('Sarah', 84), ('Sarah', 91), ('Sarah', 98), ('Tim', 120), ('Tim', 130), ('Tim', 140), ('Tim', 150), ('Ursula', 99), ('Ursula', 108), ('Ursula', 117), ('Ursula', 126), ('Victor', 99), ('Victor', 108), ('Victor', 117), ('Victor', 126), ('Wendy', 129), ('Wendy', 142), ('Wendy', 151), ('Wendy', 161), ('Xavier', 77), ('Xavier', 84), ('Xavier', 91), ('Xavier', 98), ('Yvonne', 121), ('Yvonne', 132), ('Yvonne', 143), ('Yvonne', 154), ('Zelda', 88), ('Zelda', 95), ('Zelda', 104), ('Zelda', 111)]
CPU times: user 1.39 s, sys: 127 ms, total: 1.51 s
Wall time: 8.36 s

%%time
# This one is comparatively fast and produces the same result.
from operator import add
def incr(tot, _):
    return tot + 1

result = js.foldby(key='name',
                   binop=incr,
                   initial=0,
                   combine=add,
                   combine_initial=0).compute()
print(sorted(result))

[('Alice', 700), ('Bob', 386), ('Charlie', 450), ('Dan', 337), ('Edith', 500), ('Frank', 150), ('George', 550), ('Hannah', 319), ('Ingrid', 250), ('Jerry', 750), ('Kevin', 194), ('Laura', 750), ('Michael', 691), ('Norbert', 300), ('Oliver', 494), ('Patricia', 521), ('Quinn', 671), ('Ray', 548), ('Sarah', 350), ('Tim', 540), ('Ursula', 450), ('Victor', 450), ('Wendy', 583), ('Xavier', 350), ('Yvonne', 550), ('Zelda', 398)]
CPU times: user 130 ms, sys: 21.1 ms, total: 151 ms
Wall time: 491 ms

Exercise: compute total amount per name

We want to groupby (or foldby) the name key, then add up the all of the amounts for each name.

Steps

1. Create a small function that, given a dictionary like

   {'name': 'Alice', 'transactions': [{'amount': 1, 'id': 123}, {'amount': 2, 'id': 456}]}
   

   produces the sum of the amounts, e.g. 3

2. Slightly change the binary operator of the foldby example above so that the binary operator doesn’t count the number of entries, but instead accumulates the sum of the amounts.

# Your code here...

DataFrames

For the same reasons that Pandas is often faster than pure Python, dask.dataframe can be faster than dask.bag. We will work more with DataFrames later, but from the point of view of a Bag, it is frequently the end-point of the “messy” part of data ingestion—once the data can be made into a data-frame, then complex split-apply-combine logic will become much more straight-forward and efficient.

You can transform a bag with a simple tuple or flat dictionary structure into a dask.dataframe with the to_dataframe method.

df1 = js.to_dataframe()
df1.head()

	id	name	transactions
0	0	Alice	[{'transaction-id': 17, 'amount': 179}, {'tran...
1	1	Frank	[{'transaction-id': 143, 'amount': 524}, {'tra...
2	2	Bob	[{'transaction-id': 76, 'amount': 229}, {'tran...
3	3	Patricia	[{'transaction-id': 5602, 'amount': 17}, {'tra...
4	4	Victor	[{'transaction-id': 126, 'amount': 1415}, {'tr...

This now looks like a well-defined DataFrame, and we can apply Pandas-like computations to it efficiently.

Using a Dask DataFrame, how long does it take to do our prior computation of numbers of people with the same name? It turns out that dask.dataframe.groupby() beats dask.bag.groupby() by more than an order of magnitude; but it still cannot match dask.bag.foldby() for this case.

%time df1.groupby('name').id.count().compute().head()

CPU times: user 152 ms, sys: 14.4 ms, total: 167 ms
Wall time: 1.37 s

name
Alice      700
Bob        386
Charlie    450
Dan        337
Edith      500
Name: id, dtype: int64

Denormalization

This DataFrame format is less-than-optimal because the transactions column is filled with nested data so Pandas has to revert to object dtype, which is quite slow in Pandas. Ideally we want to transform to a dataframe only after we have flattened our data so that each record is a single int, string, float, etc..

def denormalize(record):
    # returns a list for each person, one item per transaction
    return [{'id': record['id'],
             'name': record['name'],
             'amount': transaction['amount'],
             'transaction-id': transaction['transaction-id']}
            for transaction in record['transactions']]

transactions = js.map(denormalize).flatten()
transactions.take(3)

({'id': 0, 'name': 'Alice', 'amount': 179, 'transaction-id': 17},
 {'id': 0, 'name': 'Alice', 'amount': 113, 'transaction-id': 203},
 {'id': 0, 'name': 'Alice', 'amount': -15, 'transaction-id': 302})

df = transactions.to_dataframe()
df.head()

	id	name	amount	transaction-id
0	0	Alice	179	17
1	0	Alice	113	203
2	0	Alice	-15	302
3	0	Alice	204	398
4	0	Alice	39	987

%%time
# number of transactions per name
# note that the time here includes the data load and ingestion
df.groupby('name')['transaction-id'].count().compute()

CPU times: user 146 ms, sys: 21.3 ms, total: 168 ms
Wall time: 1.09 s

name
Alice       26311
Bob         28573
Charlie     18584
Dan         14083
Edith       21481
Frank        5892
George      26148
Hannah      25979
Ingrid      18419
Jerry       25661
Kevin        4597
Laura       26373
Michael     32767
Norbert     13017
Oliver      21105
Patricia    33577
Quinn       22224
Ray         20591
Sarah        7843
Tim         15113
Ursula      14489
Victor      20138
Wendy       17601
Xavier       7738
Yvonne      18477
Zelda       13219
Name: transaction-id, dtype: int64

Limitations

Bags provide very general computation (any Python function.) This generality comes at cost. Bags have the following known limitations

1. Bag operations tend to be slower than array/dataframe computations in the same way that Python tends to be slower than NumPy/Pandas

2. Bag.groupby is slow. You should try to use Bag.foldby if possible. Using Bag.foldby requires more thought. Even better, consider creating a normalised dataframe.

Learn More

• Bag documentation

• Bag screencast

• Bag API

• Bag examples

Shutdown

client.shutdown()