{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Lab 6 Reverse 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 two parallel programming assignments, which we detail on the [github repo](http://github.com/mitdbg/datascienceclass/tree/master/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.bag as db\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": [
    "## Part 1: 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 a 20x 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": [
    "**Q2. Clicker question: just by running `describe()` (and nothing else) on both `df` (pandas) and `dd_df` (dask), which of the following is true?**\n",
    "\n",
    "**a)** both output the same\n",
    "\n",
    "**b)** pandas output is more informative\n",
    "\n",
    "**c)** dask output is more informative"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So if we want dask to actually perform computations for us, we need to call `compute()`. For example, in the case of `describe()` above:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "dd_df.describe().compute()"
   ]
  },
  {
   "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",
    "**Q3. Clicker question: 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 server:"
   ]
  },
  {
   "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(1)\n",
    "\n",
    "print('\\nCluster workers:')\n",
    "client.ncores()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q4. 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": [
    "# Q4: 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 with around 7.6M rows.\n",
    "dd_df = dd.demo.make_timeseries(#start='2018-01-01',\n",
    "                                #end='2018-03-30',\n",
    "                                # NOTE: Use the smaller dataset (start_time above)\n",
    "                                # for the take home part. I'll use the earlier start time\n",
    "                                # below to obtain a larger dataset for demo'ing in class:\n",
    "                                start='2008-01-01',\n",
    "                                end='2018-03-30',\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": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print('dask shape (compute): ', dd_df.index.compute())\n",
    "print('dask shape (compute): ', len(dd_df.index))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Similarly, computing memory usage for the dataframe, as well as the number of rows in it will both call `compute()` behind the scenes:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dd_df.info(memory_usage='deep')\n",
    "print('Number of rows: ', len(dd_df))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q5. 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": [
    "# Q5: YOUR CODE GOES HERE."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, let's compare the total runtime of a sequence of operations over the dask and pandas dataframes:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# Filter and groupby using dask.\n",
    "dd_df2 = dd_df[dd_df.y > 0]\n",
    "dd_df3 = dd_df2.groupby('id').x.std()\n",
    "\n",
    "# Calling head() will also call compute() behind the scenes.\n",
    "dd_df3.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "# This gives us a pandas dataframe from the dask dataframe.\n",
    "df = dd_df.compute()\n",
    "\n",
    "# Filter and groupby using pandas.\n",
    "df2 = df[df.y > 0]\n",
    "df3 = df2.groupby('id').x.std()\n",
    "\n",
    "df3.head()\n",
    "\n",
    "# Garbage collect the pandas dataframes we created.\n",
    "#\n",
    "# XXX: pandas seems to be a lot less memory efficient than dask.\n",
    "# Even with explicit GC, perf monitors still show leaky behavior.\n",
    "# https://github.com/pandas-dev/pandas/issues?utf8=%E2%9C%93&q=is%3Aissue+leak+\n",
    "del df, df2, df3\n",
    "gc.collect()"
   ]
  },
  {
   "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['2018-01-01':'2018-02-15'].idxmax().compute()"
   ]
  },
  {
   "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",
    "df = dd_df.compute()\n",
    "df.x.rolling('1min').std().loc['2018-01-01':'2018-02-15'].idxmax()\n",
    "\n",
    "# GC pandas df.\n",
    "del df\n",
    "gc.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Reading 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 larger 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['2018-01-01':'2018-03-30']\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). You can also specify which Parquet engine to use.  There are some [`performance differences between both`](https://stackoverflow.com/questions/51361356/a-comparison-between-fastparquet-and-pyarrow), but for the time being we'll use the default (if you're using our Docker container, you have both installed and the default is `fastparquet`).\n",
    "\n",
    "**Q6. Measure the runtime of reading a dataframe from the same Parquet and CSV data we saved above. Compare using both pandas and dask, as well as fetching only column `x` vs the entire dataset.**\n",
    "\n",
    "**NOTE: When we saved our data as Parquet and CSV, we saved into multiple files (#files = #dask partitions). While dask supports reading multiple CSV files directly into a dask dataframe, pandas doesn't support the same for pandas dataframes. Take a look at [`glob`](https://docs.python.org/3/library/glob.html) for expanding the `timeseries-*.csv` into multiple files and [`pd.concat`](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.concat.html) to create a single pandas dataframe from multiple CSVs.**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Q6: YOUR CODE GOES HERE."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dask bag\n",
    "\n",
    "Next, 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. It is similar to a parallel version of itertools or a Pythonic version of the PySpark RDD.\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",
    "For example, we can look at the first 5 events published this past Monday by running:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "db.read_text('https://archive.analytics.mybinder.org/events-2019-10-28.jsonl').take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[`mybinder.org`](mybinder.org) also publishes an index containing all other JSON files that they're currently hosting:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load index JSON file, inspect its contents.\n",
    "db.read_text('https://archive.analytics.mybinder.org/index.jsonl').map(json.loads).take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using bag's [`pluck`](https://docs.dask.org/en/latest/bag-api.html#dask.bag.Bag.pluck), we can filter out for only the name attributes. We'll use that to retrieve a list of URLs of the index contents:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "urls = (db.read_text('https://archive.analytics.mybinder.org/index.jsonl')\n",
    "                    .map(json.loads)\n",
    "                    .pluck('name')\n",
    "                    .compute())\n",
    "urls = ['https://archive.analytics.mybinder.org/' + u for u in urls]\n",
    "urls[:3]"
   ]
  },
  {
   "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": [
    "notebook_runs = db.read_text(urls).map(json.loads)\n",
    "notebook_runs.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q7. Using the Python dict data we just saved above in `notebook_runs`, filter it for runs whose provider was \"GitHub\", and that happened in September 2019. 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 have two options. You can either 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)). Alternatively, you can also use `Bag`'s [`to_dataframe`](https://docs.dask.org/en/latest/bag-api.html#dask.bag.Bag.to_dataframe) and do your processing as dataframe-style computations.**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Q7: YOUR CODE GOES HERE."
   ]
  },
  {
   "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",
   "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.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}