{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Lab 6 Reversed Lecture\n",
    "\n",
    "In this reverse lecture, we'll explore in more details a subset of the functionality of [dask](dask.org), a parallel data processing library for Python. We'll spend the majority of our time on dask's [`DataFrame`](https://docs.dask.org/en/latest/dataframe.html) API, as well as examine concepts such as lazy computation and its relationship with the different schedulers.\n",
    "\n",
    "We'll follow that by data processing using columnar binary data (Parquet), and end with a brief overview of dask's [`Bag`](https://docs.dask.org/en/latest/bag.html) API, using a JSON data \"scraper\" as working example.\n",
    "\n",
    "We close with a list of additional resources for you to learn more about dask, including examples on how to use it for ML tasks. Finally, for the take home portion of this lab, you'll solve three parallel programming assignments, which we detail on the [github repo](https://github.com/mitdbg/datascienceclass/blob/master/spring_2022/lab_6/README.md). \n",
    "\n",
    "## Running this notebook\n",
    "\n",
    "To execute code from a cell, you can either click `Run` at the top, or type `shift+Enter` after clicking a cell.  You can either run the entire notebook (`Restart & Run All` from the `Kernel` drop-down), or run each cell individually.  If you choose the latter, note that it is important that you run cells in order, as later cells depend on earlier ones. And to be able to `Run All` successfully, you'll have to write code for answering some of the questions in the notebook.\n",
    "\n",
    "Once you open your notebook on the browser, and check that the cells are rendering correctly (e.g., try running the \"Python packages\" cell below), we're good to go from there."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Python packages we'll need\n",
    "\n",
    "First, import the python packages we'll be using for the lab:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "from time import perf_counter\n",
    "\n",
    "# Data processing.\n",
    "import json\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "# Q6: For loading multiple CSVs into a single (pandas) dataframe.\n",
    "import glob\n",
    "import os\n",
    "\n",
    "# Plotting.\n",
    "import matplotlib.pyplot as plt\n",
    "%matplotlib notebook\n",
    "import seaborn as sns\n",
    "sns.set_style('whitegrid')\n",
    "\n",
    "# \"Vanilla\" python parallelism.\n",
    "import multiprocessing\n",
    "\n",
    "# Scalable data analytics: dask.\n",
    "import dask\n",
    "import dask.dataframe as dd\n",
    "from dask.distributed import Client, LocalCluster\n",
    "import graphviz\n",
    "\n",
    "# Unused: scalable data analytics using Spark.\n",
    "#from pyspark.sql import SparkSession\n",
    "\n",
    "# For GC large pandas dataframes after use.\n",
    "import gc\n",
    "\n",
    "# Ignore warnings.\n",
    "import warnings\n",
    "warnings.simplefilter(\"ignore\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Intro to `dask`\n",
    "\n",
    "Dask is a python library aimed at parallel computation. It comprises two main components: a task scheduler, and a collection of data structures suited for different parallel programming tasks. Take a look at the [main page of their documentation](https://docs.dask.org/en/latest/) for more details.\n",
    "\n",
    "In this part of the reverse lecture, we'll introduce you to a subset of these features, as well as cover example tasks using the two different categories of dask schedulers.\n",
    "\n",
    "Additionally, of special interest to our course is the fact that dask provides an interface that almost exactly matches that of numpy arrays and pandas dataframes. We'll see that this is useful when speeding up some of the data analysis tasks we've covered so far in our course.\n",
    "\n",
    "\n",
    "### Example pandas vs dask comparison: data loading\n",
    "\n",
    "Before getting into too many details about the different task schedulers, let's take a look at a simple example runtime comparison between pandas and dask. Specifically, let's load the same spotify dataset from lab 4 in both, and see how long each one takes:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# load dataset using pandas:\n",
    "df = pd.read_csv('data/spotify_songs.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# load dataset using dask:\n",
    "dd_df = dd.read_csv('data/spotify_songs.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here we already see more than a 25x difference, which is impressive. However, this is also because dask does *lazy* computation.  That is, dask arrays have the required shape and data type, but they still point to data on disk. The loading is *lazy* in the sense that dask will load the array contents into memory (and in small chunks) only when necessary.\n",
    "\n",
    "We can observe this effect in action when we ask dask to compute an operation that requires inspecting the entire contents of the dataset.\n",
    "\n",
    "**Q1. Inspect the runtime each of the two dataframes (both pandas and dask) using `describe()`. How does the output in pandas compare to that of dask?**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# Q1: YOUR CODE GOES HERE (pandas)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# Q1: YOUR CODE GOES HERE (dask)    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We see that just running `describe()` on dask *still* does not perform the whole computation. If you want dask to actually perform computations, you need to call `compute()`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# Q1: YOUR CODE GOES HERE (dask)    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also inspect the task graph that dask assembles and executes for the computation:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dd_df.describe().visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we can see above, dask will try and perform as many block computations as possible, only merging and aggregating block results as needed.\n",
    "\n",
    "**Q2. Why do we have 11 \"pipelines\" in parallel for computing quantiles in the task graph above?**\n",
    "\n",
    "**a)** This is the number of workers dask is using for the computation.\n",
    "\n",
    "**b)** This is the number of categorical columns the dataframe has.\n",
    "\n",
    "**c)** This is the number of numerical columns the dataframe has.\n",
    "\n",
    "**d)** This is the number of threads dask is using for the computation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Task scheduler\n",
    "\n",
    "Dask provides two categories of task schedulers: *single machine* and *distributed*. From the client setup [documentation](https://docs.dask.org/en/latest/setup.html?highlight=client#setup): \n",
    "\n",
    "> **Single machine scheduler:** This scheduler provides basic features on a local process or thread pool. This scheduler was made first and is the default. It is simple and cheap to use. It can only be used on a single machine and does not scale.\n",
    ">\n",
    "> **Distributed scheduler:** This scheduler is more sophisticated. It offers more features, but also requires a bit more effort to set up. It can run locally or distributed across a cluster.\n",
    "\n",
    "If you load data into a dask dataframe, and run some computation on it, when you call `compute()` to start the computation it will use the *single machine scheduler*. For example, the sum over `acousticness` below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# sum values for acousticness column using pandas\n",
    "df.acousticness.sum()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# sum values for acousticness column using dask.\n",
    "# This computation uses the single machine scheduler:\n",
    "dd_df.acousticness.sum().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To use the *distributed scheduler* (even if single node on the same machine), you need to explicitly create a dask [`Client`](https://docs.dask.org/en/latest/setup.html?highlight=client#setup).  Hence, starting a dask `Client` is optional. We'll go through this step, however, as it provides an example of how to setup a [`LocalCluster`](https://distributed.dask.org/en/latest/local-cluster.html). Creating a dask client will also provide a dashboard that we can use to monitor the dask workers while they're computing results for us.\n",
    "\n",
    "There are several ways to create a dask client that connects to a `LocalCluster`. We'll do each step explicitly, although if you create a `Client` without specifying which cluster to connect to, by default dask will create a task scheduler associated to a `LocalCluster` instance.\n",
    "\n",
    "First, we check how many cores we have in our machine:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "n_cores = multiprocessing.cpu_count()\n",
    "print('number of cores we have: ', n_cores)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll use that to size the number of workers for our `LocalCluster` instance, and create a `Client` connected to it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# multithreaded:\n",
    "# cluster = LocalCluster(n_workers=1, processes=False, threads_per_worker=4)\n",
    "# multiprocessing:\n",
    "# cluster = LocalCluster(n_workers=n_cores, processes=True)\n",
    "\n",
    "# If we start out the dask-scheduler from CLI on the Docker container using:\n",
    "# $ dask-scheduler --host 0.0.0.0 --dashboard-address 8787\n",
    "#\n",
    "# Then we specify the address for the client explicitly:\n",
    "# client = Client(address='0.0.0.0:8786')\n",
    "\n",
    "# If we start out the dask-scheduler from this jupyter notebook, then need\n",
    "# to set \"ip=None\" for the status dashboard to work correctly via Docker.\n",
    "# See:\n",
    "# https://github.com/dask/distributed/issues/1875#issuecomment-387519880\n",
    "cluster = LocalCluster(ip=None, n_workers=n_cores, processes=True)\n",
    "client = Client(cluster)\n",
    "client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Info on each worker, together with how many threads per worker.\n",
    "client.ncores()\n",
    "\n",
    "# To restart the client and scheduler\n",
    "#client.restart()\n",
    "#\n",
    "# To shutdown the client and scheduler\n",
    "#client.shutdown()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can scale the cluster down:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "n_workers = n_cores / 2\n",
    "print('Scale cluster down: ')\n",
    "cluster.scale(n_workers)\n",
    "\n",
    "print('\\nWait a bit for scaling to take effect...')\n",
    "time.sleep(1)\n",
    "\n",
    "print('\\nCluster workers:')\n",
    "client.ncores()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And scale it back up again:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "n_workers = n_cores\n",
    "print('Scale cluster up again: ')\n",
    "cluster.scale(n_workers)\n",
    "\n",
    "print('\\nWait a bit for scaling to take effect...')\n",
    "time.sleep(4)\n",
    "\n",
    "print('\\nCluster workers:')\n",
    "client.ncores()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q3. Measure the time to compute the total sum of `acousticness` for `dd_df` on a local cluster with varying number of workers. Start with 1 core, and double number of cores all the way up to 8, or however many cores you have available (if less than 8). Take a look at our usage of `scale()` above for how to scale your cluster up or down, and at [`time.perf_counter`](https://docs.python.org/3/library/time.html#time.perf_counter) to measure elapsed time.** \n",
    "\n",
    "**Compare it against the time you get from using pandas. Was dask faster or slower than pandas?**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Q3: YOUR CODE GOES HERE."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we can see above, the Spotify 218k songs dataset is not large enough to justify the additional scheduler overhead from dask. Specifically, on datasets that comfortably fit within available memory, pandas is expected to be faster than dask for most operations other than loading from disk. Take a look at [\"Best Practices\"](https://docs.dask.org/en/latest/dataframe-best-practices.html) section of dask documentation for more details.\n",
    "\n",
    "So let's use a larger dataset instead. We can use dask's `demo` package to create a synthetic timeseries dataset:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This lazily creates a timeseries dataset for us.\n",
    "dd_df = dd.demo.make_timeseries(start='2021-01-01',\n",
    "                                end='2021-12-31',\n",
    "                                dtypes={'x': float, 'y': float, 'id': int},\n",
    "                                freq='1s',\n",
    "                                partition_freq='24h')\n",
    "\n",
    "dd_df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we've learned so far, dask computations are lazy up until we either explicitly call `compute()`, or call computations that themselves call `compute()`.\n",
    "\n",
    "This is why creating a large dataset above is almost instant. When we call `make_timeseries()`, calculates the number of partitions required for the parallel computation (we asked for one partition every 24h on that date range). It will only perform the actual computation required to create the dataset or brings required data chunks into memory once we actually need it to.\n",
    "\n",
    "We can see below that dask uses lazy computations even for the shape attribute:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print('pandas shape: ', df.shape)\n",
    "print('dask shape: ', dd_df.shape)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can retrieve the actual shape by again calling `compute()`, or by calling `len()`, which itself calls `compute()` behind the scenes: "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q4. Compare the runtimes of running `head()` and `len(dd_df.index)` on the dask dataframe. Which one was faster? Why do you think that was the case?**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Q4: YOUR CODE GOES HERE."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Operations over the index is where dask usually does a lot better than pandas (see [\"Best Practices\"](https://docs.dask.org/en/latest/dataframe-best-practices.html)):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# Dask: rolling standard deviation of 'x' column on 1m windows over the Datetime index.\n",
    "# Return index of first occurrence of max value out of those rolling 1m stddev.\n",
    "dd_df.x.rolling('1min').std().loc['2021-01-01':'2021-02-15'].idxmax().compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# Get a Pandas dataframe out of dask.\n",
    "df = dd_df.compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# Same operation in Pandas\n",
    "df.x.rolling('1min').std().loc['2021-01-01':'2021-02-15'].idxmax()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# GC pandas df\n",
    "del df\n",
    "gc.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Writing and loading data\n",
    "\n",
    "Let's take a look at processing data in column-oriented binary format using [`Parquet`](https://en.wikipedia.org/wiki/Apache_Parquet). As we've seen in class, Parquet is an open-source column-oriented data storage format. In addition to allowing us to store our dataframe data in a column-wise fashion, it also supports a number of compression techniques (e.g., dictionary encoding, RLE, and bit packing, as we've seen in class).\n",
    "\n",
    "In this part of the lab, we'll briefly look at how to store and read data written in Parquet format, and compare it to loading the same dataset from CSV.\n",
    "\n",
    "First, we can save the timeseries dataset as both Parquet and CSV format. We'll do this for a smaller subset of the data, as it takes a really long time to save the entire dataset as CSV."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "subset_df = dd_df.loc['2021-01-01':'2021-1-31']\n",
    "\n",
    "# Measure time for saving as Parquet files.\n",
    "t1_start = perf_counter()\n",
    "subset_df.to_parquet('data/parquet/')\n",
    "t1_stop = perf_counter()\n",
    "print('to_parquet() (ms): %s' % ((t1_stop - t1_start)*1000))\n",
    "\n",
    "# Measure time for saving as CSV files.\n",
    "t1_start = perf_counter()\n",
    "subset_df.to_csv('data/csv/timeseries-*.csv')\n",
    "t1_stop = perf_counter()\n",
    "print('to_csv() (ms): %s' % ((t1_stop - t1_start)*1000))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Even though we used dask to write to Parquet, both pandas and dask support writing to / reading from Parquet files. The method interface is the same: [pandas's `read_parquet`](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.read_parquet.html) vs [dask's `read_parquet`](https://docs.dask.org/en/latest/dataframe-api.html#dask.dataframe.read_parquet).\n",
    "\n",
    "**Q5. Measure the runtime of reading a dataframe from the same Parquet and CSV data we saved above using `dask`. Compare to reading the Parquet file using `pandas`.**\n",
    "\n",
    "**NOTE: Pandas does not support reading csv with a file pattern. While you could use the `pd.concat` function, you do not need to benchmark `pd.read_csv`**."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Q5: YOUR CODE GOES HERE.\n",
    "expt_name = \"...\"\n",
    "t1_start = perf_counter()\n",
    "df = None\n",
    "t1_stop = perf_counter()\n",
    "print(f'{expt_name} (ms): {((t1_stop - t1_start)*1000)}')\n",
    "del df\n",
    "gc.collect()\n",
    "\n",
    "expt_name = \"...\"\n",
    "..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dask bag\n",
    "\n",
    "Finally, we'll look at processing JSON data hosted on the web using the dask [`Bag`](https://docs.dask.org/en/latest/bag.html) data structure ([full API](https://docs.dask.org/en/latest/bag-api.html)). \n",
    "\n",
    "There is some overlap between what you can do with bags and dataframes, as we'll see below. However, bags are useful when the data you're analyzing is more naturally represented as Python objects (e.g., dicts) than as tabular data (dataframe). That's often the case with JSON data, which has an almost 1:1 mapping with Python dicts.\n",
    "\n",
    "From the official documentation:\n",
    "\n",
    ">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.\n",
    "\n",
    "\n",
    "### Example: using Bag to process JSON data\n",
    "\n",
    "Below we'll use events JSON data from a web service that runs Jupyter notebooks called [mybinder.org](http://mybinder.org). Every time a user launches a notebook on platforms such as GitHub or GitLab, mybinder publishes an event and stores it in publicly accessible JSON files (one per day).\n",
    "\n",
    "Let's go through a step by step example of bags. Suppose we want to get information about notebooks published in 2022."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.bag as db"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[`mybinder.org`](mybinder.org) publishes an index containing the JSON files that they're currently hosting. To read it, we will use the `read_text` function which lazily returns a bag containing the list of lines of text. The `take(n)` function will return `n` elements of the bag."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load index file, inspect its contents.\n",
    "events = db.read_text('https://archive.analytics.mybinder.org/index.jsonl')\n",
    "events.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As you can see, each line represents json text about daily events. We want to convert json text to python dictionaries. We will use the `map()` function. `map(f)` applies the `f` function to every element in the bag. We should thus call `map(json.loads)`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Turn lines text to python dicts.\n",
    "events = events.map(json.loads)\n",
    "events.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's now filter by date. We will use the `filter(p)` function which apply the `p` predicate to every item, and returns only the items that satisfy the predicate."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Filter by year.\n",
    "events = events.filter(lambda entry: '2022' in entry['date'])\n",
    "events.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's extract the filename. We will use the `pluck(key)` function, which will extract the `key` from the dictionary. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get the names.\n",
    "events = events.pluck(\"name\")\n",
    "events.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using a bag on the list of urls, we can automatically parse the JSON data into Python dict:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Fetch the urls.\n",
    "urls = events.map(lambda name: f\"https://archive.analytics.mybinder.org/{name}\")\n",
    "notebook_runs = db.read_text(urls.compute()).map(json.loads)\n",
    "notebook_runs.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q6. Using the Python dict data we just saved above in `notebook_runs`, filter it for runs whose provider was \"GitHub\", and that happened in March 2022. What were the top-3 most run notebooks in that month, and how many times were they run?**\n",
    "\n",
    "**To answer this question, you should use only `Bag` functions, such as [`filter`](https://docs.dask.org/en/latest/bag-api.html#dask.bag.Bag.filter) and [`frequencies`](https://docs.dask.org/en/latest/bag-api.html#dask.bag.Bag.frequencies) (take a look at some [usage examples here](https://examples.dask.org/bag.html)).**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Q6: YOUR CODE GOES HERE.\n",
    "# Hint: Use the 'spec' attribute for the notebook name."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Naturally, the top 3 are all jupyter tutorials."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Additional Resources\n",
    "\n",
    "## Dask examples and tutorials\n",
    "\n",
    "* Dask examples: https://examples.dask.org/\n",
    "* Dask for ML tasks (e.g., incremental learning and hyperparameter tuning): https://ml.dask.org/examples.html\n",
    "* YouTube playlist with introductory Dask concepts: https://www.youtube.com/playlist?list=PLTgRMOcmRb3OlkfAdqJWyGGrQM7eU-mi7\n",
    "\n",
    "## Dask benchmark codes\n",
    "* On a 512 core server: https://matthewrocklin.com/blog/work/2017/07/03/scaling\n",
    "* On GCS: https://gist.github.com/mrocklin/4c198b13e92f881161ef175810c7f6bc#file-scaling-gcs-ipynb\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}