lab6_take_home_Q2_Q3.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "from time import perf_counter\n",
    "\n",
    "# Data processing.\n",
    "import json\n",
    "import numpy as np\n",
    "\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": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "from collections import defaultdict"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "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-02-01',\n",
    "                                # Use the one above; I'll use this larger one 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",
    "pd_df = dd_df.compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [],
   "source": [
    "def scale(cores):\n",
    "    n_workers = min(cores, multiprocessing.cpu_count())\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",
    "    print(client.ncores())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table style=\"border: 2px solid white;\">\n",
       "<tr>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3 style=\"text-align: left;\">Client</h3>\n",
       "<ul style=\"text-align: left; list-style: none; margin: 0; padding: 0;\">\n",
       "  <li><b>Scheduler: </b>tcp://127.0.0.1:33867</li>\n",
       "  <li><b>Dashboard: </b><a href='http://127.0.0.1:42547/status' target='_blank'>http://127.0.0.1:42547/status</a>\n",
       "</ul>\n",
       "</td>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3 style=\"text-align: left;\">Cluster</h3>\n",
       "<ul style=\"text-align: left; list-style:none; margin: 0; padding: 0;\">\n",
       "  <li><b>Workers: </b>8</li>\n",
       "  <li><b>Cores: </b>8</li>\n",
       "  <li><b>Memory: </b>10.46 GB</li>\n",
       "</ul>\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "<Client: 'tcp://127.0.0.1:33867' processes=8 threads=8, memory=10.46 GB>"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "n_workers = 8\n",
    "cluster = LocalCluster(ip=None, n_workers=8, processes=True, threads_per_worker=1)\n",
    "client = Client(cluster)\n",
    "client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_q2_results():\n",
    "    window_5min = 60*5\n",
    "    pd_times = []\n",
    "    dd_times = []\n",
    "    cores = [1,2,4,6,8]\n",
    "    for i in cores:\n",
    "        scale(i)\n",
    "        # Get dask results\n",
    "        start_dd = perf_counter()\n",
    "        dd_result = dd_df.y.rolling(window=window_5min).mean().compute()\n",
    "        dd_times.append((perf_counter() - start_dd)*1000)\n",
    "        # Get pandas results\n",
    "        start_pd = perf_counter()\n",
    "        pd_result = pd_df.y.rolling(window=window_5min).mean()\n",
    "        pd_times.append((perf_counter() - start_pd)*1000)\n",
    "    return dd_times, dd_result, pd_times, pd_result\n",
    "\n",
    "q2_results = get_q2_results()\n",
    "\n",
    "# Plot\n",
    "cores = [1,2,4,6,8]\n",
    "width = 0.4\n",
    "plt.bar([c+width/2 for c in cores], q2_results[0], width, label=\"dask times\")\n",
    "plt.bar([c-width/2 for c in cores], q2_results[2], width, label=\"pandas times\")\n",
    "plt.ylabel(\"Time (ms)\")\n",
    "plt.xlabel(\"Cores\")\n",
    "plt.legend()\n",
    "plt.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "({'name': 'events-2018-11-03.jsonl', 'date': '2018-11-03', 'count': '7057'},\n",
       " {'name': 'events-2018-11-04.jsonl', 'date': '2018-11-04', 'count': '7489'},\n",
       " {'name': 'events-2018-11-05.jsonl', 'date': '2018-11-05', 'count': '13590'},\n",
       " {'name': 'events-2018-11-06.jsonl', 'date': '2018-11-06', 'count': '13920'},\n",
       " {'name': 'events-2018-11-07.jsonl', 'date': '2018-11-07', 'count': '12766'},\n",
       " {'name': 'events-2018-11-08.jsonl', 'date': '2018-11-08', 'count': '14105'},\n",
       " {'name': 'events-2018-11-09.jsonl', 'date': '2018-11-09', 'count': '11843'},\n",
       " {'name': 'events-2018-11-10.jsonl', 'date': '2018-11-10', 'count': '7047'},\n",
       " {'name': 'events-2018-11-11.jsonl', 'date': '2018-11-11', 'count': '6940'},\n",
       " {'name': 'events-2018-11-12.jsonl', 'date': '2018-11-12', 'count': '16322'},\n",
       " {'name': 'events-2018-11-13.jsonl', 'date': '2018-11-13', 'count': '16530'},\n",
       " {'name': 'events-2018-11-14.jsonl', 'date': '2018-11-14', 'count': '14099'},\n",
       " {'name': 'events-2018-11-15.jsonl', 'date': '2018-11-15', 'count': '13182'},\n",
       " {'name': 'events-2018-11-16.jsonl', 'date': '2018-11-16', 'count': '12863'},\n",
       " {'name': 'events-2018-11-17.jsonl', 'date': '2018-11-17', 'count': '6490'},\n",
       " {'name': 'events-2018-11-18.jsonl', 'date': '2018-11-18', 'count': '7310'},\n",
       " {'name': 'events-2018-11-19.jsonl', 'date': '2018-11-19', 'count': '13348'},\n",
       " {'name': 'events-2018-11-20.jsonl', 'date': '2018-11-20', 'count': '13982'},\n",
       " {'name': 'events-2018-11-21.jsonl', 'date': '2018-11-21', 'count': '13165'},\n",
       " {'name': 'events-2018-11-22.jsonl', 'date': '2018-11-22', 'count': '12217'})"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "## Q3\n",
    "mybinder_url = 'https://archive.analytics.mybinder.org/'\n",
    "db.read_text(mybinder_url+\"index.jsonl\").map(json.loads).take(20)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_q3_results():\n",
    "    \n",
    "    urls = (db.read_text('https://archive.analytics.mybinder.org/index.jsonl')\n",
    "                        .map(json.loads)\n",
    "                        .pluck('name')\n",
    "                        .filter(lambda x: x.split('-')[2] == \"08\" and x.split('-')[1] == \"2019\")\n",
    "                        .compute())\n",
    "    urls = ['https://archive.analytics.mybinder.org/' + u for u in urls]\n",
    "    results = defaultdict(lambda: 0)\n",
    "    for url in urls:\n",
    "        data = db.read_text(url).map(json.loads)\n",
    "        for d in data:\n",
    "            results[d[\"provider\"]] += 1\n",
    "    return results\n",
    "\n",
    "\n",
    "## Q3 \n",
    "q3_times = []\n",
    "cores = [1,2,4]\n",
    "repetitions = 5\n",
    "for num_cores in cores:\n",
    "    tot_time = 0\n",
    "    scale(num_cores)\n",
    "    for i in range(repetitions):\n",
    "        start = perf_counter()\n",
    "        get_q3_results()\n",
    "        tot_time += perf_counter() - start\n",
    "    q3_times.append((tot_time/repetitions)*1000)\n",
    "\n",
    "    \n",
    "# Plot Timing Results\n",
    "plt.bar(cores, q3_times)\n",
    "# plt.plot(cores, q2_results[2], label=\"pandas times\")\n",
    "plt.ylabel(\"Time (ms)\")\n",
    "plt.xlabel(\"Cores\")\n",
    "plt.show()\n",
    "\n",
    "# Sort Results\n",
    "results_list = list()\n",
    "for key, value in results.items():\n",
    "    results_list.append((key, value))\n",
    "results_list = sorted(results_list, key=lambda x: x[1], reverse=True)\n",
    "\n",
    "results_list"
   ]
  }
 ],
 "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
}