03-Extra.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Extra Demos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client, LocalCluster\n",
    "\n",
    "cluster = LocalCluster(n_workers=2, threads_per_worker=1, memory_limit='700MiB')\n",
    "client = Client(cluster)\n",
    "client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.dataframe as ddf\n",
    "\n",
    "raw = ddf.read_parquet('data/checkouts-micro')\n",
    "\n",
    "raw"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Activity 1: Publishers\n",
    "\n",
    "What are the top 50 publishers in the Seattle library system by...\n",
    "* checkout activity (easier)\n",
    "* library material holdings (harder)\n",
    "\n",
    "Hints:\n",
    "* Try to use Dask's `nlargest` or `nsmallest` for ordered results with a limit (like 50).\n",
    "    * That approach is vastly more efficient than trying to sort a big dataset.\n",
    "* For top publishers by library holdings...\n",
    "    * the same item may appear in many months of data\n",
    "    * Pandas/Dask doesn't have the same \"COUNT DISTINCT\" operator as SQL so you may have to get a bit creative\n",
    "    * if you don't narrow (hint!) down the data, it will be hard to run this query with the allocated cluster resources"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q1 = raw.groupby('Publisher').agg({'Checkouts':'sum'})\n",
    "q1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q1.nlargest(50, 'Checkouts').compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q2 = raw[['Publisher', 'Title']].drop_duplicates().groupby('Publisher').agg({'Publisher':'count'})\n",
    "q2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q2.nlargest(50, 'Publisher').compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Activity 2: Popular subjects\n",
    "\n",
    "*Bonus Project*\n",
    "\n",
    "Notice that the Subjects field contains a string list of subjects.\n",
    "\n",
    "If we want to analyze checkouts by subject, we might start by trying to parse this field into a Python list. Like Pandas, Dask allows us to split strings as well as explode collections into multiple rows.\n",
    "\n",
    "Try to find the top 10 subjects by checkout activity. Hint: Try to eliminate as much data as you can from the dataset as early as possible."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "narrowed = raw[['Checkouts', 'Subjects']]\n",
    "narrowed.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "narrowed.Subjects.str.split(', ').head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "narrowed['Subject'] = narrowed.Subjects.str.split(', ')\n",
    "narrowed = narrowed.drop(columns='Subjects')\n",
    "narrowed.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "narrowed.explode('Subject').head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can check that the basic logic works..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "narrowed.explode('Subject').groupby('Subject').agg({'Checkouts':'sum'}).head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "But for large datasets you might want to narrow down the volume of data earlier -- i.e., the query above runs on the full amount of data and the `head` call can only limits the rows that come back from the result because the limit can't be pushed through the aggregation. \n",
    "\n",
    "Another way to \"test\" the query is to pull, say, two partitions:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "narrowed.partitions[:1].explode('Subject').groupby('Subject').agg({'Checkouts':'sum'}).head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "narrowed.explode('Subject').groupby('Subject').agg({'Checkouts':'sum'}).nlargest(10, 'Checkouts').compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.close()\n",
    "cluster.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.8.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}