<jupyter_start><jupyter_text>Table of Contents1&nbsp;&nbsp;Bag basics1.1&nbsp;&nbsp;Getting the data and creating the bag1.2&nbsp;&nbsp;persist1.3&nbsp;&nbsp;filter1.4&nbsp;&nbsp;map1.5&nbsp;&nbsp;Using map and predefined functions1.6&nbsp;&nbsp;flatten1.7&nbsp;&nbsp;compute   Bag basics [Introduction to Dask, by <NAME>](https://github.com/andersy005/dask-notebooks)This notebook will introduce basic but essential `Dask.Bag` operations. At the same time we will introduce different concepts such as peristence in Dask. Getting the data and creating the bagAs we did in our first notebook, we will use the reduced dataset (10 percent) provided for the KDD Cup 1999, containing nearly half million network interactions. The file is provided as a Gzip file that we will download locally.If you previously downloaded this dataset, the following program won't download it again.<jupyter_code>import urllib.request
import os
import pathlib
DATASET_DIR = pathlib.Path('../data')
DATASET_FILEPATH = pathlib.Path('../data/kddcup.data_10_percent.gz')
if not DATASET_DIR.exists():
    os.mkdir(DATA_DIR)
    f = urllib.request.urlretrieve(
        "http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz",
        DATASET_FILEPATH)<jupyter_output><empty_output><jupyter_text>Now we can use this file to create our dask bag.<jupyter_code>import dask
import dask.bag as db
# Progress Bar
from dask.diagnostics import ProgressBar
pbar = ProgressBar()
pbar.register()
raw_data = db.read_text(DATASET_FILEPATH)
# Take a look at a few elements
raw_data.take(2)
raw_data<jupyter_output><empty_output><jupyter_text>`persist`It is often ideal to load, filter, and shuffle data once and keep this result in memory. Afterwards each of the several complex queries can be based off of this in-memory data rather than have to repeat the full load-filter-shuffle process each time. To do this, we can use `.persist()` method to tell dask to persist the data in memory.<jupyter_code>raw_data.persist()<jupyter_output>[########################################] | 100% Completed | 11.6s<jupyter_text>```filter```This method can be applied to bags in order to keep just elements that satisfy a certain condition. More concretely, a function is evaluated on every element in the original bag. The new resulting bag will contain just those elements that make the function return True.For example, imagine we want to count how many normal. interactions we have in our dataset. We can filter our raw_data bag as follows.<jupyter_code>normal_raw_data = raw_data.filter(lambda x: 'normal' in x)
normal_count = normal_raw_data.count()
print("There are {} 'normal' interactions".format(normal_count.compute()))
print("There are {} 'total' interactions".format(raw_data.count().compute()))<jupyter_output>[########################################] | 100% Completed |  0.8s
There are 494021 'total' interactions<jupyter_text>We have a total of 494021 in our 10 percent dataset. Here we can see that 97278 contain the `normal.` tag word.  `map` By using the map method, we can apply a function to every element in our bag. Python's lambdas are specially expressive for this particular.In this case we want to read our data file as a CSV formatted one. We can do this by applying a lambda function to each element in the bag as follows.<jupyter_code>csv_data = raw_data.map(lambda x: x.split(","))
from pprint import pprint
pprint(csv_data.take(3))<jupyter_output>[########################################] | 100% Completed |  0.1s
(['0',
  'tcp',
  'http',
  'SF',
  '181',
  '5450',
  '0',
  '0',
  '0',
  '0',
  '0',
  '1',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '8',
  '8',
  '0.00',
  '0.00',
  '0.00',
  '0.00',
  '1.00',
  '0.00',
  '0.00',
  '9',
  '9',
  '1.00',
  '0.00',
  '0.11',
  '0.00',
  '0.00',
  '0.00',
  '0.00',
  '0.00',
  'normal.\n'],
 ['0',
  'tcp',
  'http',
  'SF',
  '239',
  '486',
  '0',
  '0',
  '0',
  '0',
  '0',
  '1',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '8',
  '8',
  '0.00',
  '0.00',
  '0.00',
  '0.00',
  '1.00',
  '0.00',
  '0.00',
  '19',
  '19',
  '1.00',
  '0.00',
  '0.05',
  '0.00',
  '0.00',
  '0.00',
  '0.00',
  '0.00',
  'normal.\n'],
 ['0',
  'tcp',
  'http',
  'SF',
  '235',
  '1337',
  '0',
  '0',
  '0',
  '0',
  '0',
  '1',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '0',
  '8',
  '8',
  '0.00',
  '0.00',
  '0.00',
  '0[...]<jupyter_text>Again, dask operations are lazily evaluate, i.e the computation happens once we call the dask to do so (i.e. take in this case). What if we take a lot of elements instead of just the first few?<jupyter_code>samples = csv_data.take(10000)<jupyter_output>[########################################] | 100% Completed |  6.2s<jupyter_text>We can see that it takes longer. The map function is applied now in a distributed way to a lot of elements in the bag, hence the longer execution time.  Using `map` and predefined functionsOf course we can use predefined functions with map. Imagine we want to have each element in the bag as a key-value pair where the key is the tag (e.g. normal) and the value is the whole list of elements that represents the row in the CSV formatted file. We could proceed as follows:<jupyter_code>def parse_interaction(line):
    elems = line.split(",")
    tag = elems[41][:-2]
    return (tag, elems)
key_csv_data = raw_data.map(parse_interaction)
pprint(key_csv_data.take(3))<jupyter_output>[########################################] | 100% Completed |  0.1s
(('normal',
  ['0',
   'tcp',
   'http',
   'SF',
   '181',
   '5450',
   '0',
   '0',
   '0',
   '0',
   '0',
   '1',
   '0',
   '0',
   '0',
   '0',
   '0',
   '0',
   '0',
   '0',
   '0',
   '0',
   '8',
   '8',
   '0.00',
   '0.00',
   '0.00',
   '0.00',
   '1.00',
   '0.00',
   '0.00',
   '9',
   '9',
   '1.00',
   '0.00',
   '0.11',
   '0.00',
   '0.00',
   '0.00',
   '0.00',
   '0.00',
   'normal.\n']),
 ('normal',
  ['0',
   'tcp',
   'http',
   'SF',
   '239',
   '486',
   '0',
   '0',
   '0',
   '0',
   '0',
   '1',
   '0',
   '0',
   '0',
   '0',
   '0',
   '0',
   '0',
   '0',
   '0',
   '0',
   '8',
   '8',
   '0.00',
   '0.00',
   '0.00',
   '0.00',
   '1.00',
   '0.00',
   '0.00',
   '19',
   '19',
   '1.00',
   '0.00',
   '0.05',
   '0.00',
   '0.00',
   '0.00',
   '0.00',
   '0.00',
   'normal.\n']),
 ('normal',
  ['0',
   'tcp',
   'http',
   'SF',
   '235',
   '1337',
   '0',
   '0',
   '0',
   '0',
[...]<jupyter_text>That was easy, wasn't it?In our notebook about working with key-value pairs we will use this type of bag to do data aggregations (e.g. count by key)  `flatten`This method concatenate nested lists into one long list.<jupyter_code>raw_data.flatten().count().compute()<jupyter_output>[########################################] | 100% Completed |  7.2s<jupyter_text>`compute``compute` method computes the dask collection it is applied to.  Basically it will get all the elements in the lazily evaluate bag into memory for us to work with them. For this reason it has to be used with care, specially when working with large bags.<jupyter_code>raw_data.compute()[:10]<jupyter_output>[########################################] | 100% Completed | 34.2s<jupyter_text>As a last example combining all the previous, we want to compute all the normal interactions as key-value pairs.<jupyter_code># get data from file and repartition it in 8 partitions and pe
raw_data = db.read_text(DATASET_FILEPATH).repartition(8)

# parse into key-value pairs
key_csv_data = raw_data.map(parse_interaction)

# filter normal key interactions and persist them 
normal_key_interactions = key_csv_data.filter(lambda x: x[0] == "normal")

# Compute all
all_normal = normal_key_interactions.compute()
print(len(all_normal))<jupyter_output>97278<jupyter_text>This count matches with the previous count for normal interactions. The new procedure is more time consuming. This is because we retrieve all the data with compute and then use Python's len on the resulting list. Before we were just counting the total number of elements in the bag by using count. NOTE: **At its core, dask is a graph execution engine. It allows us to construct a prescription for the computation we want to carry out.**As a result, we can use `.visualize()` method to render the computation of an object’s task graph using graphviz.<jupyter_code>normal_key_interactions.visualize()
%load_ext version_information
%version_information dask<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.17770198753381158, 'fraction_numerical': 0.05280489239092085, 'mean_word_length': 3.766816143497758, 'pattern_counts': {'":': 0, '<': 38, '<?xml version=': 0, '>': 38, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '151709', 'license': 'MIT', 'n_tokens_mistral': 3258, 'n_tokens_neox': 3058, 'n_words': 984, 'path': '01-dask-bags/02-bag-basics.ipynb', 'repo_name': 'andersy005/dask-notebooks', 'size': '8515'}