dask-tutorial-notebook.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Dask Tutorial\n",
    "### By Michelle Lam, Elke Windschitl, & Michael Zargari for EDS-217"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is a tutoriol on how to use the Python library Dask. Dask is a tool to scale data libraries in Python such as Numpy, Pandas, and Scikit-learn. This means that smaller libraries can be scaled to used on big data. Dask can be deployed anywhere, so users can start on a laptop and scale up to cloud computing.\n",
    "\n",
    "Visit https://www.dask.org/ to learn more about Dask.\n",
    "\n",
    "## Why use Dask?\n",
    "Dask can be used when working with big data sets. Environmental data scientists may encounter big data sets frequently. But why do we need to use dask? \n",
    "\n",
    "*\"Big data is data sets that are so voluminous and complex that traditional data processing application software are inadequate to deal with them.\"* (Wikipedia)\n",
    "\n",
    "Libraries such as Numpy and Pandas aren't built to handle big data sets. Dask works with Numpy and Pandas under the hood to run familiar functions on large sets of data. The maintainers of Dask are the same maintainers of Numpy and Pandas.\n",
    "\n",
    "<img src=\"./dask-screenshot.png\" width = \"600\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## How does Dask work?\n",
    "\n",
    "Arrays, Dataframes, and semi-structured/unstructured data take up a great deal of RAM when processed directly with Pandas, Numpy, and Scikit-learn. These functions are not great at memory management so they front-load their work which puts stress on your RAM and causes it to slow down when fed in too much data.\n",
    "\n",
    "<img src=\"./Example-Structured-Unstructured-Semi-Structured-Data-e1638423518970.jpg\"/>\n",
    "\n",
    "Dask resolves this by efficiently breaking down your data and feeding it to the RAM in bite sized pieces, ultimately combining it back into readable and usable data once complete.\n",
    "\n",
    "Dask has three main \"collections\" called Dask Array, Dask Dataframe, and Dask-ML (Dask Bag) which are alternatives to Pandas DataFrame, Numpy Array and Scikit-learn. \n",
    "\n",
    "Below, we will be descibing how Dask manages each of these collections to make them more efficient.\n",
    "\n",
    "<br>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3 Different Parts to the Dask project\n",
    "### 1. Dask Collections (\"core-library\")\n",
    "- **High-level collections**: mimic NumPy, lists, and pandas, but can operate in parallel on datasets that don't fit into memory\n",
    "    - Array\n",
    "    - Bag\n",
    "    - DataFrame\n",
    "- **Low-level collections**: give you finer control in building custom parallel and distributed computations\n",
    "    - Delayed\n",
    "    - Futures\n",
    "\n",
    "### 2. Dask Cluster\n",
    "Dask uses a distributed scheduler, which exists in the context of a Dask cluster.\n",
    "\n",
    "Structure of a dask cluster:\n",
    "\n",
    "<img src=\"./dask_cluster_img.png\" width = \"600\"/>\n",
    "\n",
    "### 3. Dask Ecosystem\n",
    "The Dask ecosystem connects several adiitional open source projects that provide different mechanisms for deploying Dask clusters. \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## High Level Collection\n",
    "### **Dask Array**\n",
    "A Dask array is made up of smaller Numpy arrays that are fed into an algorithm to allow computation on arrays that are larger than your available RAM memory. During an Array operation, Dask translates the array operation into a task graph which breaks up large Numpy arrays into multiple smaller chunks, and executes the work on each chunk at the same time. This is called parallel computing. Results from each chunk are combined to produce the final output. \n",
    "\n",
    "### **Dask Dataframe**\n",
    "A Dask DataFrame maintains the familiar Pandas code structure and language, making it easy for Pandas users to scale up DataFrame workloads, just replace ```pd``` with ```dd```. During a DataFrame operation, Dask creates a task graph and breaks down the dataframe into smaller parts that reduces memory footprint and increases RAM efficiency by sharing and deleting intermediate results while computing.\n",
    "\n",
    "### **Dask-ML (Dask Bag)**\n",
    "Dask Bag is used on semi-structured/unstructured data such as JSON records, text data, log files, or user-defined Python objects using operations such as filter, fold, map and groupby. This allows you to do things such as hefty text analysis and data scraping that would be difficult without Dask's proper resource management. \n",
    "\n",
    "**This tutorial will focus on using the high-level collection of DataFrame.**\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Getting Started\n",
    "To get started using Dask in Python, check to see if it is already on your laptop. Dask is included with Anaconda, so it may have installed when you downloaded Anaconda.\n",
    "\n",
    "```\n",
    "# To check for dask, try importing. \n",
    "import dask\n",
    "```\n",
    "\n",
    "If you do not have dask, Python will let you know. If you get a message saying 'No module named dask', you will need to install dask using conda in the command line before you import.\n",
    "\n",
    "```\n",
    "# To install dask use conda in your Powershell command line:\n",
    "conda install dask\n",
    "```\n",
    "<div class=\"run\">\n",
    "    ▶️ <b> Run the cell(s) below. </b>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# To get all of Dask run:\n",
    "import dask\n",
    "\n",
    "# To use Dask Data Frames run:\n",
    "import dask.dataframe as dd\n",
    "\n",
    "# To time things\n",
    "import time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As mentioned above, Dask is made to handle large data sets. However, for the purposes of this class, we will be using a smaller data set as an example. \n",
    "\n",
    "Here we use data from: *Ecological and social Interactions in urban parks: bird surveys in local parks in the central Arizona-Phoenix metropolitan area* by Warren, Paige S.; Research Assistant Professor; University of Massachusetts-Amherst\n",
    "Kinzig, Ann; Assistant Professor; Arizona State University\n",
    "Martin, Chris A; Associate Professor; Arizona State University\n",
    "Machabee, Louis; Global Institute of Sustainability, Arizona State University\n",
    "\n",
    "<div class=\"run\">\n",
    "    ▶️ <b> Run the cell(s) below. </b>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create an object that is a link to data\n",
    "birds_link = 'https://portal.edirepository.org/nis/dataviewer?packageid=knb-lter-cap.256.10&entityid=53edaa7a0e083013d9bf20322db1780e'\n",
    "\n",
    "# Read in the data using dd.read_csv()\n",
    "birds_dask = dd.read_csv(birds_link)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Compare with Pandas\n",
    "Notice that ```dd.read_csv()``` is similar to ```pd.read_csv()```. Functions that are used in Pandas can be used in Dask. Compare reading in a data frame in dask with reading in a data frame with Pandas.\n",
    "\n",
    "<div class=\"run\">\n",
    "    ▶️ <b> Run the cell(s) below. </b>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import Pandas\n",
    "import pandas as pd\n",
    "\n",
    "# Create the same data frame using pandas\n",
    "birds_pandas = pd.read_csv(birds_link)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Look at you pandas output\n",
    "print(type(birds_pandas))\n",
    "print('...........................................................................')\n",
    "print(birds_pandas.head)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now compare these outputs with your dask data frame. What is different?\n",
    "<div class=\"run\">\n",
    "    ▶️ <b> Run the cell(s) below. </b>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(type(birds_dask))\n",
    "print('...........................................................................')\n",
    "print(birds_dask.head)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Pandas version of the data frame is `<class 'pandas.core.frame.DataFrame'>` while the Dask version of the dataframe is `<class 'dask.dataframe.core.DataFrame'>`. The Pandas data frame shows up when you look at it, but the Dask data frame appears empty... why? Dask expects to be receiving a huge data set, so it does not automatically display the rows to prevent crashing. From the Dask documentation: \n",
    "\n",
    "\"Unlike Pandas, Dask DataFrames are lazy, meaning that data is only loaded when it is needed for a computation. No data is printed here, instead it is replaced by ellipses (...).\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using Dask functions and methods \n",
    "### (Hint: these are the same as in Pandas!)\n",
    "Explore the outputs of the code and consider why they look the way they do\n",
    "<div class=\"run\">\n",
    "    ▶️ <b> Run the cell(s) below. </b>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create small subset of data (first 50 rows) in Dask\n",
    "small_birds_dask = birds_dask.loc[:50]\n",
    "\n",
    "#find the mean of a column in Dask\n",
    "print(birds_dask.bird_count.mean())\n",
    "\n",
    "#add a column to the dask data frame\n",
    "birds_dask['bird_count_add'] = birds_dask.bird_count + 1\n",
    "print(birds_dask.info)\n",
    "\n",
    "#applying functions - create a new column that calculates the seen to heard ratio of birds\n",
    "def seen_calc(seen):\n",
    "    \"\"\" Multiplies the seen value by 100\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "        seen : float\n",
    "            Number of birds seen\n",
    "       \n",
    "    Returns\n",
    "    -------\n",
    "        seen_new : float\n",
    "            Number of birds seen times 100\n",
    "    \"\"\"\n",
    "    seen_new = (seen * 100)\n",
    "    return seen_new\n",
    "\n",
    "birds_dask['seen_new'] = birds_dask.seen\n",
    "birds_dask.seen.apply(seen_calc, meta=('seen', 'int64'))\n",
    "print(birds_dask.info)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using Dask on big data\n",
    "Now that we've looked at Dask, lets try it out with some bigger data. Here, we artifically create a large data frame by copying rows of the same data frame. Try to use Pandas and see what happens. Try to use Dask and see what happens.\n",
    "<div class=\"run\">\n",
    "    ▶️ <b> Run the cell(s) below. </b>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Make a large Pandas data frame\n",
    "start_time = time.time()\n",
    "\n",
    "birds_pandasx2 = pd.concat([birds_pandas, birds_pandas])\n",
    "\n",
    "birds_pandasx4 = pd.concat([birds_pandasx2, birds_pandasx2])\n",
    "\n",
    "birds_pandasx8 = pd.concat([birds_pandasx4, birds_pandasx4])\n",
    "\n",
    "birds_pandasx16 = pd.concat([birds_pandasx8, birds_pandasx8])\n",
    "\n",
    "birds_pandasx32 = pd.concat([birds_pandasx16, birds_pandasx16])\n",
    "\n",
    "birds_pandasx64 = pd.concat([birds_pandasx32, birds_pandasx32])\n",
    "\n",
    "birds_pandasx128 = pd.concat([birds_pandasx64, birds_pandasx64])\n",
    "\n",
    "birds_pandasx256 = pd.concat([birds_pandasx128, birds_pandasx128])\n",
    "\n",
    "birds_pandasx512 = pd.concat([birds_pandasx256, birds_pandasx256])\n",
    "\n",
    "birds_pandasx1024 = pd.concat([birds_pandasx512, birds_pandasx512])\n",
    "\n",
    "birds_pandasx2048 = pd.concat([birds_pandasx1024, birds_pandasx1024])\n",
    "\n",
    "birds_pandasx4096 = pd.concat([birds_pandasx2048, birds_pandasx2048])\n",
    "\n",
    "end_time = time.time()\n",
    "\n",
    "print(f\"It took {end_time-start_time} seconds\")\n",
    "\n",
    "print(birds_pandasx4096.head)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Make a large Dask data frame\n",
    "start_time = time.time()\n",
    "\n",
    "birds_daskx2 = dd.multi.concat([birds_dask, birds_dask])\n",
    "\n",
    "birds_daskx4 = dd.multi.concat([birds_daskx2, birds_daskx2])\n",
    "\n",
    "birds_daskx8 = dd.multi.concat([birds_daskx4, birds_daskx4])\n",
    "\n",
    "birds_daskx16 = dd.multi.concat([birds_daskx8, birds_daskx8])\n",
    "\n",
    "birds_daskx32 = dd.multi.concat([birds_daskx16, birds_daskx16])\n",
    "\n",
    "birds_daskx64 = dd.multi.concat([birds_daskx32, birds_daskx32])\n",
    "\n",
    "birds_daskx128 = dd.multi.concat([birds_daskx64, birds_daskx64])\n",
    "\n",
    "birds_daskx256 = dd.multi.concat([birds_daskx128, birds_daskx128])\n",
    "\n",
    "birds_daskx512 = dd.multi.concat([birds_daskx256, birds_daskx256])\n",
    "\n",
    "birds_daskx1024 = dd.multi.concat([birds_daskx512, birds_daskx512])\n",
    "\n",
    "birds_daskx2048 = dd.multi.concat([birds_daskx1024, birds_daskx1024])\n",
    "\n",
    "birds_daskx4096 = dd.multi.concat([birds_daskx2048, birds_daskx2048])\n",
    "\n",
    "#Run a calculation in Dask\n",
    "birds_daskx4096['distancex100'] = birds_daskx4096['distance']*100\n",
    "\n",
    "end_time = time.time()\n",
    "\n",
    "print(f\"It took {end_time-start_time} seconds\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# WARNING: BREAK YOUR CODE\n",
    "## ONLY RUN THE FOLLOWING IF YOU WANT TO SEE PANDAS CRASH\n",
    "### It may take some time to get your code running again. Proceed at your own risk.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Attempt a calculation with Pandas\n",
    "birds_pandasx4096['distancex100'] = birds_pandasx4096['distance']*100"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Who uses Dask? What can it be used for?\n",
    "\n",
    "\n",
    "\n",
    "### *Life sciences*\n",
    "\n",
    "**Harvard Medical School, Howard Hughes Medical Institute, Chan Zuckerberg Initiative, and the UC Berkeley Advanced Bioimaging Center** all use Dask for high resolution, 4-dimensional, cellular imagery. This generates large amounts of data that is difficult to analyze with traditional methods. \n",
    "- Dask helps them scale their data analysis workflows with its familiar API that resembles NumPy, Pandas, and Scikit-learn code. \n",
    "\n",
    "Dask is also used at the Novartis Institute for Biomedical Research to scale machine learning prototypes.\n",
    "\n",
    "<br>\n",
    "\n",
    "### *Geophysical sciences*\n",
    "\n",
    "Dask is used in Climate Science, Energy, Hydrology, Meteorology, and Satellite Imaging by companies such as NASA, LANL, PANGEO, and the UK Meteorology Office.\n",
    "\n",
    "- With Dask, oceanographers can produce massive simulated datasets of the earth’s oceans and researchers can look at large seismology datasets from sensors around the world, collect a large number of observations from satellites and weather stations, and run big simulations.\n",
    "\n",
    "<br>\n",
    "\n",
    "### *Corporations*\n",
    "**Walmart** uses Dask for forecasting the demand for 500,000,000 store-item combinations. To provide in-demand items in sufficient quantities at all their outlets, they need to run huge computations. Using RAPIDS and XGBoost, supported by Dask, they reached 100x acceleration.\n",
    "\n",
    "**Blue Yonder** uses Dask to process terabytes of data on a daily basis. They can write Pandas-like code in Dask, which can then be pushed directly to production. This helps keep their feedback cycles short and waste low.\n",
    "\n",
    "**Capital One** uses Dask to accelerate their machine learning pipelines.\n",
    "\n",
    "**Barclays** usesd Dask for financial system modeling."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## References\n",
    "\n",
    "Warren, P.S., A. Kinzig, C.A. Martin, and L. Machabee. 2021. Ecological and social Interactions in urban parks: bird surveys in local parks in the central Arizona-Phoenix metropolitan area ver 10. Environmental Data Initiative. https://doi.org/10.6073/pasta/f6f004bc7112ce266fde2b80fad19ff4 (Accessed 2022-09-08).\n",
    "\n",
    "https://en.wikipedia.org/wiki/Dask_(software)#High-Level_collections\n",
    "\n",
    "https://stories.dask.org/en/latest/\n",
    "\n",
    "https://earth-env-data-science.github.io/lectures/dask/dask_arrays.html\n",
    "\n",
    "https://earth-env-data-science.github.io/lectures/dask/intro.html\n",
    "\n",
    "https://en.wikipedia.org/wiki/Big_data\n",
    "\n",
    "https://tutorial.dask.org/00_overview.html"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.10.4 ('eds217')",
   "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.10.4"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "01298427d7509dc791212f5580d72e2a69e14f859fb9b9009471feff7d08282d"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}