09 Scaling - Dask.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Dask\n",
    "\n",
    "## Introduction\n",
    "Pandas is a great tool and is used for a variety of purposes like data cleansing, exploratory data analysis, time series analysis, visual analysis, building features for ML models to name a few. We <3 pandas! However, we have seen that as soon as you hit scale, things start slowing down. People generally switch to Spark Data Frames. Porting pandas to spark DFs can be painful and might not be efficient until and unless you have super large datasets.\n",
    "\n",
    "One of our data engineering piece started off small (100s of thousand data points / day) but quickly became a largish data problem (5M rows / day). That is when we decided to try out dask. Our chunked pandas dataframe techniques used to take 20 hrs to perform complex data cleansing on 300k rows vs 2.5 million rows in 60 mins on dask!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## What is Dask?\n",
    "Dask provides a framework for performing parallel computing for analytics.\n",
    "\n",
    "Dask is composed of two components:\n",
    "\n",
    "* Dynamic task scheduling optimized for computation. This is similar to Airflow, Luigi, Celery, or Make, but optimized for interactive computational workloads.\n",
    "* “Big Data” collections like parallel arrays, dataframes, and lists that extend common interfaces like NumPy, Pandas, or Python iterators to larger-than-memory or distributed environments. These parallel collections run on top of the dynamic task schedulers."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Broad Categories of Dask Components\n",
    "\n",
    "* **Dask DataFrame** - mimics Pandas\n",
    "* **Dask Bag** - mimics iterators, Toolz, and PySpark\n",
    "* **Dask Delayed** - mimics for loops and wraps custom code"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dask Dataframe\n",
    "\n",
    "Dask dataframe is constituted of multiple pandas dataframes split along an index. The smaller pandas dataframes may reside in memory or on disk (if it does not fit in memory) or on multiple machines (in case of dask cluster)\n",
    "\n",
    "![](./img/dask-dataframe.png)\n",
    "\n",
    "Because the dask.dataframe application programming interface (API) is a subset of the Pandas API it should be familiar to Pandas users. There are some slight alterations due to the parallel nature of dask."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Let's try it out"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.dataframe as dd\n",
    "import pandas as pd\n",
    "%matplotlib inline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Connect to SQL Table"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create SQL alchemy engine"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask._version import get_versions\n",
    "get_versions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sqlalchemy import create_engine\n",
    "import sqlite3\n",
    "import numpy as np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "engine = create_engine('sqlite:///datasets/database.sqlite')\n",
    "conn = sqlite3.connect('datasets/database.sqlite')\n",
    "uri = 'sqlite:///datasets/database.sqlite'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pdf = pd.read_sql(\"select * from status limit 10000000\", con=engine)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pdf.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pdf.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf = dd.from_pandas(pdf, npartitions=16)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.multiprocessing import get"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "pdf.apply(\n",
    "        lambda x: x['bikes_available']**2 * x['docks_available']**2,\n",
    "    axis=1\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "ddf.apply(\n",
    "        lambda x: x['bikes_available']**2 * x['docks_available']**2,\n",
    "    axis=1\n",
    "    ).compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf.apply(\n",
    "        lambda x: x['bikes_available']**2 * x['docks_available']**2,\n",
    "    axis=1\n",
    "    ).visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## read sql directly in Dask"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "meta = {\n",
    "#     'id': np.int,\n",
    "      'station_id': np.int,\n",
    "    'bikes_available': np.int,\n",
    "    'docks_available': np.int,\n",
    "    'time': np.datetime64\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data= {\n",
    "#     'id': 0,\n",
    "    'station_id': 1,\n",
    "    'bikes_available': 2,\n",
    "    'docks_available': 1,\n",
    "    'time': '2014-08-30 12:01:20'\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cols = ['station_id','bikes_available','docks_available','time']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = pd.DataFrame(columns=cols)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for c in df.columns:\n",
    "    df[c] = df[c].astype(meta[c])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf = dd.read_sql_table(\"status\", uri=uri, index_col='station_id', npartitions=16)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf._meta"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "type(ddf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Shuffling - Why index is important"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "ddf.groupby(['station_id'])['docks_available'].apply(lambda x: max(x)).compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "ddf.groupby(['bikes_available'])['docks_available'].apply(lambda x: max(x)).compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf.groupby(['station_id'])['docks_available'].apply(lambda x: max(x)).visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf.groupby(['bikes_available'])['docks_available'].apply(lambda x: max(x)).visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First approach needs a lot of shuffling. Now imagine if you deployed the same code on a cluster. Shuffling here means network IO. This can slow down the computation very much"
   ]
  }
 ],
 "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.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}