dask-tools.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "d25f7303-a717-46df-9c0a-f5428fb8ea8c",
   "metadata": {},
   "source": [
    "<img src=\"assets/dask_horizontal.svg\"\n",
    "     width=\"45%\"\n",
    "     alt=\"Dask logo\\\">\n",
    "     \n",
    "# Parallel Computing in Python with Dask\n",
    "\n",
    "This notebook provides a high-level overview of Dask. We discuss why you might want to use Dask, the high-level and low-level APIs for generating computational graphs, and the schedulers which allow for the parallel execution of these graphs."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "76dee750-423f-4d1c-9576-f4545938e496",
   "metadata": {},
   "source": [
    "# Overview\n",
    "\n",
    "### What is Dask?\n",
    "\n",
    "- Dask is a flexible, open source library for parallel computing in Python\n",
    "\n",
    "    - GitHub: [https://github.com/dask/dask](https://github.com/dask/dask)\n",
    "    \n",
    "    - Documentation: [https://docs.dask.org](https://docs.dask.org)\n",
    "\n",
    "- Scales the existing Python ecosystem\n",
    "\n",
    "### Why Dask?\n",
    "\n",
    "- Enables parallel and larger-than-memory computations\n",
    "- Uses familiar APIs you're used to from projects like NumPy, pandas, and scikit-learn\n",
    "- Allows you to scale existing workflows with minimal code changes\n",
    "- Dask works on your laptop, but also scales out to large clusters\n",
    "- Offers great built-in diagnosic tools\n",
    "\n",
    "### Dask vs. Spark\n",
    "\n",
    "#### Reasons you might choose Spark\n",
    "\n",
    "- You prefer Scala or the SQL language\n",
    "- You have mostly JVM infrastructure and legacy systems\n",
    "- You want an established and trusted solution for business\n",
    "- You are mostly doing business analytics with some lightweight machine learning\n",
    "- You want an all-in-one solution\n",
    "\n",
    "#### Reasons you might choose Dask\n",
    "\n",
    "- You prefer Python or native code, or have large legacy code bases that you do not want to entirely rewrite\n",
    "- Your use case is complex or does not cleanly fit the Spark computing model\n",
    "- You want a lighter-weight transition from local computing to cluster computing\n",
    "- You want to interoperate with other technologies and don’t mind installing multiple packages\n",
    "\n",
    "### Components of Dask\n",
    "\n",
    "<img src=\"assets/dask-overview.png\"\n",
    "     width=\"85%\"\n",
    "     alt=\"Dask components\\\">\n",
    "     "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9244937f-67d9-4e91-8e16-fb3ac9cb6886",
   "metadata": {},
   "source": [
    "# Parallel and Distributed Machine Learning\n",
    "\n",
    "[Dask-ML](https://dask-ml.readthedocs.io) has resources for parallel and distributed machine learning.\n",
    "\n",
    "### Types of Scaling\n",
    "\n",
    "There are a couple of distinct scaling problems you might face.\n",
    "The scaling strategy depends on which problem you're facing.\n",
    "\n",
    "1. CPU-Bound: Data fits in RAM, but training takes too long. Many hyperparameter combinations, a large ensemble of many models, etc.\n",
    "2. Memory-bound: Data is larger than RAM, and sampling isn't an option.\n",
    "\n",
    "![](assets/ml-dimensions.png)\n",
    "\n",
    "* For in-memory problems, just use scikit-learn (or your favorite ML library).\n",
    "* For large models, use `dask_ml.joblib` and your favorite scikit-learn estimator\n",
    "* For large datasets, use `dask_ml` estimators"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "77854303-33cb-460c-b123-0282ff5620fd",
   "metadata": {},
   "source": [
    "### Machine Learning with sklearn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "e7d55c36-400e-4787-9591-d34dba5d524b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.905\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "array([[-0.77244139,  0.3607576 , -2.38110133,  0.08757   ],\n",
       "       [ 1.14946035,  0.62254594,  0.37302939,  0.45965795],\n",
       "       [-0.77694695,  0.31434299, -2.26231851,  0.06339125],\n",
       "       [ 0.79010037,  0.68530624, -0.44740487,  0.44692959]])"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from sklearn.datasets import make_classification\n",
    "from sklearn.svm import SVC\n",
    "from sklearn.model_selection import GridSearchCV\n",
    "\n",
    "X, y = make_classification(n_samples=10000, n_features=4, random_state=0)\n",
    "estimator = SVC(random_state=0)\n",
    "estimator.fit(X, y)\n",
    "print(estimator.score(X, y))\n",
    "estimator.support_vectors_[:4]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "929fe438-032d-4d54-a047-5c63968a61d6",
   "metadata": {},
   "source": [
    "### Hyperparameter Optimization\n",
    "\n",
    "There are a few ways to learn the best *hyper*parameters while training. One is `GridSearchCV`.\n",
    "As the name implies, this does a brute-force search over a grid of hyperparameter combinations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "2b1ce06d-c7ef-4c2c-8120-bea1e2f28578",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Fitting 2 folds for each of 4 candidates, totalling 8 fits\n",
      "[CV] END ................................C=0.001, kernel=rbf; total time=   8.4s\n",
      "[CV] END ................................C=0.001, kernel=rbf; total time=   9.9s\n",
      "[CV] END ...............................C=0.001, kernel=poly; total time=   3.7s\n",
      "[CV] END ...............................C=0.001, kernel=poly; total time=   3.8s\n",
      "[CV] END .................................C=10.0, kernel=rbf; total time=   2.9s\n",
      "[CV] END .................................C=10.0, kernel=rbf; total time=   2.8s\n",
      "[CV] END ................................C=10.0, kernel=poly; total time=   3.5s\n",
      "[CV] END ................................C=10.0, kernel=poly; total time=   3.3s\n",
      "CPU times: user 46.5 s, sys: 348 ms, total: 46.9 s\n",
      "Wall time: 47 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "({'C': 10.0, 'kernel': 'rbf'}, 0.9086000000000001)"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "estimator = SVC(gamma='auto', random_state=0, probability=True)\n",
    "param_grid = {\n",
    "    'C': [0.001, 10.0],\n",
    "    'kernel': ['rbf', 'poly'],\n",
    "}\n",
    "\n",
    "grid_search = GridSearchCV(estimator, param_grid, verbose=2, cv=2)\n",
    "grid_search.fit(X, y)\n",
    "grid_search.best_params_, grid_search.best_score_"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "852551cd-cb86-42ff-86b2-1a94f0783655",
   "metadata": {},
   "source": [
    "### Single-machine parallelism with scikit-learn\n",
    "\n",
    "![](assets/unmerged_grid_search_graph.svg)\n",
    "\n",
    "Scikit-Learn has nice *single-machine* parallelism, via Joblib.\n",
    "Any scikit-learn estimator that can operate in parallel exposes an `n_jobs` keyword.\n",
    "This controls the number of CPU cores that will be used."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "377be5dd-9456-426e-b516-5a6ca7d94f80",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Fitting 2 folds for each of 4 candidates, totalling 8 fits\n",
      "CPU times: user 9.3 s, sys: 250 ms, total: 9.55 s\n",
      "Wall time: 28.9 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "({'C': 10.0, 'kernel': 'rbf'}, 0.9086000000000001)"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "grid_search = GridSearchCV(estimator, param_grid, verbose=2, cv=2, n_jobs=-1)\n",
    "grid_search.fit(X, y)\n",
    "grid_search.best_params_, grid_search.best_score_"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c1146cf4-04f7-45c0-85a3-bb4f963bdd4a",
   "metadata": {},
   "source": [
    "### Multi-machine parallelism with Dask\n",
    "\n",
    "![](assets/merged_grid_search_graph.svg)\n",
    "\n",
    "Dask can talk to scikit-learn (via joblib) so that your *cluster* is used to train a model. \n",
    "\n",
    "If you run this on a laptop, it will take quite some time, but the CPU usage will be satisfyingly near 100% for the duration. To run faster, you would need a disrtibuted cluster. That would mean putting something in the call to `Client` something like\n",
    "\n",
    "```\n",
    "c = Client('tcp://my.scheduler.address:8786')\n",
    "```\n",
    "\n",
    "Details on the many ways to create a cluster can be found [here](https://docs.dask.org/en/latest/setup/single-distributed.html)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "687c7ef3-aaa2-49bf-bf5d-5118a5335aaa",
   "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:49198</li>\n",
       "  <li><b>Dashboard: </b><a href='http://127.0.0.1:8787/status' target='_blank'>http://127.0.0.1:8787/status</a></li>\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>4</li>\n",
       "  <li><b>Cores: </b>16</li>\n",
       "  <li><b>Memory: </b>16.00 GiB</li>\n",
       "</ul>\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "<Client: 'tcp://127.0.0.1:49198' processes=4 threads=16, memory=16.00 GiB>"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from dask.distributed import Client\n",
    "import joblib\n",
    "\n",
    "client = Client(n_workers=4)\n",
    "client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "65219e3f-1c8d-4bfd-ad7f-da781322518d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Fitting 2 folds for each of 4 candidates, totalling 8 fits\n",
      "CPU times: user 10.3 s, sys: 337 ms, total: 10.7 s\n",
      "Wall time: 27.1 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "({'C': 10.0, 'kernel': 'rbf'}, 0.9086000000000001)"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "estimator = SVC(gamma='auto', random_state=0, probability=True)\n",
    "param_grid = {\n",
    "    'C': [0.001, 10.0],\n",
    "    'kernel': ['rbf', 'poly'],\n",
    "}\n",
    "\n",
    "grid_search = GridSearchCV(estimator, param_grid, verbose=2, cv=2, n_jobs=-1)\n",
    "\n",
    "with joblib.parallel_backend(\"dask\", scatter=[X, y]):\n",
    "    grid_search.fit(X, y)\n",
    "    \n",
    "grid_search.best_params_, grid_search.best_score_"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4442d5b2-40ad-4933-9217-3ad50729dd69",
   "metadata": {},
   "source": [
    "### Training on Large Datasets\n",
    "\n",
    "Sometimes you'll want to train on a larger than memory dataset. `dask-ml` has implemented estimators that work well on dask arrays and dataframes that may be larger than your machine's RAM."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "fcd7d59f-dfff-47be-b5cf-78601943c4ee",
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.array as da\n",
    "import dask.delayed\n",
    "from sklearn.datasets import make_blobs\n",
    "import numpy as np"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e1052dec-afad-4880-bd32-1cdab85522e7",
   "metadata": {},
   "source": [
    "We'll make a small (random) dataset locally using scikit-learn."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "bede99a9-668c-4c55-b157-1339a2cc1b8c",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([[ 1.00796679,  4.34582168,  2.15175661,  1.04337835, -1.82115164,\n",
       "         2.81149666, -1.18757701,  7.74628882,  9.36761449, -2.20570731,\n",
       "         5.71142324,  0.41084221,  1.34168817,  8.4568751 , -8.59042755,\n",
       "        -8.35194302, -9.55383028,  6.68605157,  5.34481483,  7.35044606],\n",
       "       [ 9.49283024,  6.1422784 , -0.97484846,  5.8604399 , -7.61126963,\n",
       "         2.86555735, -7.25390288,  8.89609285,  0.33510318, -1.79181328,\n",
       "        -4.66192239,  5.43323887, -0.86162507,  1.3705568 , -9.7904172 ,\n",
       "         2.3613231 ,  2.20516237,  2.20604823,  8.76464833,  3.47795068],\n",
       "       [-2.67206588, -1.30103177,  3.98418492, -8.88040428,  3.27735964,\n",
       "         3.51616445, -5.81395151, -7.42287114, -3.73476887, -2.89520363,\n",
       "         1.49435043, -1.35811028,  9.91250767, -7.86133474, -5.78975793,\n",
       "        -6.54897163,  3.08083281, -5.18975209, -0.85563107, -5.06615534],\n",
       "       [-6.85980599, -7.87144648,  3.33572279, -7.00394241, -5.97224874,\n",
       "        -2.55638942,  6.36329802, -7.97988653,  6.80059611, -8.14552537,\n",
       "         9.48255539, -0.67232114,  9.38462699,  2.09067352,  4.80505419,\n",
       "        -9.14866204, -4.32240399, -7.61670696, -4.14166466, -7.73998277]])"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "n_centers = 12\n",
    "n_features = 20\n",
    "\n",
    "X_small, y_small = make_blobs(n_samples=1000, centers=n_centers, n_features=n_features, random_state=0)\n",
    "\n",
    "centers = np.zeros((n_centers, n_features))\n",
    "\n",
    "for i in range(n_centers):\n",
    "    centers[i] = X_small[y_small == i].mean(0)\n",
    "    \n",
    "centers[:4]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "65c4e9cb-a624-4bad-bb0f-0a8381777c05",
   "metadata": {},
   "source": [
    "The small dataset will be the template for our large random dataset.\n",
    "We'll use `dask.delayed` to adapt `sklearn.datasets.make_blobs`, so that the actual dataset is being generated on our workers. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "fa8876a8-e038-4c66-a4ee-e12e4b018b70",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr>\n",
       "<td>\n",
       "<table>\n",
       "  <thead>\n",
       "    <tr><td> </td><th> Array </th><th> Chunk </th></tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr><th> Bytes </th><td> 2.98 GiB </td> <td> 30.52 MiB </td></tr>\n",
       "    <tr><th> Shape </th><td> (20000000, 20) </td> <td> (200000, 20) </td></tr>\n",
       "    <tr><th> Count </th><td> 400 Tasks </td><td> 100 Chunks </td></tr>\n",
       "    <tr><th> Type </th><td> float64 </td><td> numpy.ndarray </td></tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</td>\n",
       "<td>\n",
       "<svg width=\"75\" height=\"170\" style=\"stroke:rgb(0,0,0);stroke-width:1\" >\n",
       "\n",
       "  <!-- Horizontal lines -->\n",
       "  <line x1=\"0\" y1=\"0\" x2=\"25\" y2=\"0\" style=\"stroke-width:2\" />\n",
       "  <line x1=\"0\" y1=\"6\" x2=\"25\" y2=\"6\" />\n",
       "  <line x1=\"0\" y1=\"12\" x2=\"25\" y2=\"12\" />\n",
       "  <line x1=\"0\" y1=\"18\" x2=\"25\" y2=\"18\" />\n",
       "  <line x1=\"0\" y1=\"25\" x2=\"25\" y2=\"25\" />\n",
       "  <line x1=\"0\" y1=\"31\" x2=\"25\" y2=\"31\" />\n",
       "  <line x1=\"0\" y1=\"37\" x2=\"25\" y2=\"37\" />\n",
       "  <line x1=\"0\" y1=\"43\" x2=\"25\" y2=\"43\" />\n",
       "  <line x1=\"0\" y1=\"50\" x2=\"25\" y2=\"50\" />\n",
       "  <line x1=\"0\" y1=\"56\" x2=\"25\" y2=\"56\" />\n",
       "  <line x1=\"0\" y1=\"62\" x2=\"25\" y2=\"62\" />\n",
       "  <line x1=\"0\" y1=\"68\" x2=\"25\" y2=\"68\" />\n",
       "  <line x1=\"0\" y1=\"75\" x2=\"25\" y2=\"75\" />\n",
       "  <line x1=\"0\" y1=\"81\" x2=\"25\" y2=\"81\" />\n",
       "  <line x1=\"0\" y1=\"87\" x2=\"25\" y2=\"87\" />\n",
       "  <line x1=\"0\" y1=\"93\" x2=\"25\" y2=\"93\" />\n",
       "  <line x1=\"0\" y1=\"100\" x2=\"25\" y2=\"100\" />\n",
       "  <line x1=\"0\" y1=\"106\" x2=\"25\" y2=\"106\" />\n",
       "  <line x1=\"0\" y1=\"112\" x2=\"25\" y2=\"112\" />\n",
       "  <line x1=\"0\" y1=\"120\" x2=\"25\" y2=\"120\" style=\"stroke-width:2\" />\n",
       "\n",
       "  <!-- Vertical lines -->\n",
       "  <line x1=\"0\" y1=\"0\" x2=\"0\" y2=\"120\" style=\"stroke-width:2\" />\n",
       "  <line x1=\"25\" y1=\"0\" x2=\"25\" y2=\"120\" style=\"stroke-width:2\" />\n",
       "\n",
       "  <!-- Colored Rectangle -->\n",
       "  <polygon points=\"0.0,0.0 25.412616514582485,0.0 25.412616514582485,120.0 0.0,120.0\" style=\"fill:#8B4903A0;stroke-width:0\"/>\n",
       "\n",
       "  <!-- Text -->\n",
       "  <text x=\"12.706308\" y=\"140.000000\" font-size=\"1.0rem\" font-weight=\"100\" text-anchor=\"middle\" >20</text>\n",
       "  <text x=\"45.412617\" y=\"60.000000\" font-size=\"1.0rem\" font-weight=\"100\" text-anchor=\"middle\" transform=\"rotate(-90,45.412617,60.000000)\">20000000</text>\n",
       "</svg>\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "dask.array<concatenate, shape=(20000000, 20), dtype=float64, chunksize=(200000, 20), chunktype=numpy.ndarray>"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "n_samples_per_block = 200000\n",
    "n_blocks = 100\n",
    "\n",
    "delayeds = [dask.delayed(make_blobs)(n_samples=n_samples_per_block,\n",
    "                                     centers=centers,\n",
    "                                     n_features=n_features,\n",
    "                                     random_state=i)[0]\n",
    "            for i in range(n_blocks)]\n",
    "arrays = [da.from_delayed(obj, shape=(n_samples_per_block, n_features), dtype=X.dtype)\n",
    "          for obj in delayeds]\n",
    "X = da.concatenate(arrays)\n",
    "X"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "d5bb7282-b3dc-461c-a9f5-33d1b234cc9c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# X = X.persist()  # Only run this on the cluster."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f182fafa-15be-44f9-97e9-713ba3577cb7",
   "metadata": {},
   "source": [
    "The algorithms implemented in Dask-ML are scalable. They handle larger-than-memory datasets just fine.\n",
    "\n",
    "They follow the scikit-learn API, so if you're familiar with scikit-learn, you'll feel at home with Dask-ML."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "1f3e66af-9028-4075-a3de-88333374eb08",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 16.4 s, sys: 2.23 s, total: 18.6 s\n",
      "Wall time: 1min 29s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "array([0, 1, 0, 3, 3, 3, 4, 1, 3, 7], dtype=int32)"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "from dask_ml.cluster import KMeans\n",
    "\n",
    "clf = KMeans(init_max_iter=3, oversampling_factor=10)\n",
    "clf.fit(X)\n",
    "clf.labels_[:10].compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "190a0707-3d73-4988-8a5e-1c5c498fe2a0",
   "metadata": {},
   "outputs": [],
   "source": [
    "client.shutdown()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3114d00e-2534-48d6-8d5a-98c558edb39b",
   "metadata": {},
   "source": [
    "# Dask-SQL\n",
    "\n",
    "Core features\n",
    "\n",
    "* SQL parsing, optimization, planning, translation for Dask\n",
    "* Start with data from...\n",
    "    * files in the cloud (e.g., S3)\n",
    "    * any data in Python (e.g., Pandas or Dask Dataframe)\n",
    "    * modern data catalog/aggregation like Intake (https://github.com/intake/intake)\n",
    "    * __direct from enterprise data lakes/warehouses: Hive Metastore, Databricks, etc.__\n",
    "        * Bring the SQL integration power of Spark right into the Python/Dask world\n",
    "* Query cached datasets to leverage the speed of a large distributed memory pool\n",
    "\n",
    "Bonus features\n",
    "* user-defined functions\n",
    "* a SQL server\n",
    "* ML in SQL\n",
    "* a command-line client\n",
    "* more in the works!\n",
    "\n",
    "Learn more...\n",
    "* Homepage: https://nils-braun.github.io/dask-sql/\n",
    "* Docs: https://dask-sql.readthedocs.io/en/latest/\n",
    "* Source: https://github.com/nils-braun/dask-sql"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "ba6d8c27-32f0-48fb-bc0a-31cb7b1b684c",
   "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:51435</li>\n",
       "  <li><b>Dashboard: </b><a href='http://127.0.0.1:8787/status' target='_blank'>http://127.0.0.1:8787/status</a></li>\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>4</li>\n",
       "  <li><b>Cores: </b>16</li>\n",
       "  <li><b>Memory: </b>16.00 GiB</li>\n",
       "</ul>\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "<Client: 'tcp://127.0.0.1:51435' processes=4 threads=16, memory=16.00 GiB>"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from dask.distributed import Client\n",
    "\n",
    "client = Client()\n",
    "client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "2d0b2ded-b52b-4218-b60f-7382fdf14a72",
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask_sql import Context\n",
    "\n",
    "c = Context()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "281f7286-7d50-4fc5-928a-2fe16251fcfe",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><strong>Dask DataFrame Structure:</strong></div>\n",
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Date</th>\n",
       "      <th>DayOfWeek</th>\n",
       "      <th>DepTime</th>\n",
       "      <th>CRSDepTime</th>\n",
       "      <th>ArrTime</th>\n",
       "      <th>CRSArrTime</th>\n",
       "      <th>UniqueCarrier</th>\n",
       "      <th>FlightNum</th>\n",
       "      <th>TailNum</th>\n",
       "      <th>ActualElapsedTime</th>\n",
       "      <th>CRSElapsedTime</th>\n",
       "      <th>AirTime</th>\n",
       "      <th>ArrDelay</th>\n",
       "      <th>DepDelay</th>\n",
       "      <th>Origin</th>\n",
       "      <th>Dest</th>\n",
       "      <th>Distance</th>\n",
       "      <th>TaxiIn</th>\n",
       "      <th>TaxiOut</th>\n",
       "      <th>Cancelled</th>\n",
       "      <th>Diverted</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>npartitions=10</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>datetime64[ns]</td>\n",
       "      <td>int64</td>\n",
       "      <td>float64</td>\n",
       "      <td>int64</td>\n",
       "      <td>float64</td>\n",
       "      <td>int64</td>\n",
       "      <td>object</td>\n",
       "      <td>int64</td>\n",
       "      <td>object</td>\n",
       "      <td>float64</td>\n",
       "      <td>float64</td>\n",
       "      <td>float64</td>\n",
       "      <td>float64</td>\n",
       "      <td>float64</td>\n",
       "      <td>object</td>\n",
       "      <td>object</td>\n",
       "      <td>float64</td>\n",
       "      <td>float64</td>\n",
       "      <td>float64</td>\n",
       "      <td>bool</td>\n",
       "      <td>int64</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>\n",
       "<div>Dask Name: read-csv, 10 tasks</div>"
      ],
      "text/plain": [
       "Dask DataFrame Structure:\n",
       "                          Date DayOfWeek  DepTime CRSDepTime  ArrTime CRSArrTime UniqueCarrier FlightNum TailNum ActualElapsedTime CRSElapsedTime  AirTime ArrDelay DepDelay  Origin    Dest Distance   TaxiIn  TaxiOut Cancelled Diverted\n",
       "npartitions=10                                                                                                                                                                                                                            \n",
       "                datetime64[ns]     int64  float64      int64  float64      int64        object     int64  object           float64        float64  float64  float64  float64  object  object  float64  float64  float64      bool    int64\n",
       "                           ...       ...      ...        ...      ...        ...           ...       ...     ...               ...            ...      ...      ...      ...     ...     ...      ...      ...      ...       ...      ...\n",
       "...                        ...       ...      ...        ...      ...        ...           ...       ...     ...               ...            ...      ...      ...      ...     ...     ...      ...      ...      ...       ...      ...\n",
       "                           ...       ...      ...        ...      ...        ...           ...       ...     ...               ...            ...      ...      ...      ...     ...     ...      ...      ...      ...       ...      ...\n",
       "                           ...       ...      ...        ...      ...        ...           ...       ...     ...               ...            ...      ...      ...      ...     ...     ...      ...      ...      ...       ...      ...\n",
       "Dask Name: read-csv, 10 tasks"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import dask.dataframe as dd\n",
    "import os\n",
    "\n",
    "df = dd.read_csv(os.path.join('data', 'nycflights', '*.csv'),\n",
    "                 parse_dates={'Date': [0, 1, 2]},\n",
    "                 dtype={'TailNum': str,\n",
    "                        'CRSElapsedTime': float,\n",
    "                        'Cancelled': bool})\n",
    "\n",
    "df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "fd93e4c9-fd21-4fbf-85fb-71072b321bd1",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><strong>Dask DataFrame Structure:</strong></div>\n",
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Date</th>\n",
       "      <th>DayOfWeek</th>\n",
       "      <th>DepTime</th>\n",
       "      <th>CRSDepTime</th>\n",
       "      <th>ArrTime</th>\n",
       "      <th>CRSArrTime</th>\n",
       "      <th>UniqueCarrier</th>\n",
       "      <th>FlightNum</th>\n",
       "      <th>TailNum</th>\n",
       "      <th>ActualElapsedTime</th>\n",
       "      <th>CRSElapsedTime</th>\n",
       "      <th>AirTime</th>\n",
       "      <th>ArrDelay</th>\n",
       "      <th>DepDelay</th>\n",
       "      <th>Origin</th>\n",
       "      <th>Dest</th>\n",
       "      <th>Distance</th>\n",
       "      <th>TaxiIn</th>\n",
       "      <th>TaxiOut</th>\n",
       "      <th>Cancelled</th>\n",
       "      <th>Diverted</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>npartitions=1</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>datetime64[ns]</td>\n",
       "      <td>int64</td>\n",
       "      <td>float64</td>\n",
       "      <td>int64</td>\n",
       "      <td>float64</td>\n",
       "      <td>int64</td>\n",
       "      <td>object</td>\n",
       "      <td>int64</td>\n",
       "      <td>object</td>\n",
       "      <td>float64</td>\n",
       "      <td>float64</td>\n",
       "      <td>float64</td>\n",
       "      <td>float64</td>\n",
       "      <td>float64</td>\n",
       "      <td>object</td>\n",
       "      <td>object</td>\n",
       "      <td>float64</td>\n",
       "      <td>float64</td>\n",
       "      <td>float64</td>\n",
       "      <td>bool</td>\n",
       "      <td>int64</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>\n",
       "<div>Dask Name: getitem, 34 tasks</div>"
      ],
      "text/plain": [
       "Dask DataFrame Structure:\n",
       "                         Date DayOfWeek  DepTime CRSDepTime  ArrTime CRSArrTime UniqueCarrier FlightNum TailNum ActualElapsedTime CRSElapsedTime  AirTime ArrDelay DepDelay  Origin    Dest Distance   TaxiIn  TaxiOut Cancelled Diverted\n",
       "npartitions=1                                                                                                                                                                                                                            \n",
       "               datetime64[ns]     int64  float64      int64  float64      int64        object     int64  object           float64        float64  float64  float64  float64  object  object  float64  float64  float64      bool    int64\n",
       "                          ...       ...      ...        ...      ...        ...           ...       ...     ...               ...            ...      ...      ...      ...     ...     ...      ...      ...      ...       ...      ...\n",
       "Dask Name: getitem, 34 tasks"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "c.create_table(\"flights\", df)\n",
    "\n",
    "result = c.sql('SELECT * FROM flights LIMIT 10')\n",
    "result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "1f13b7c4-58d1-4390-ab6f-f04dddd45b8a",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dask.dataframe.core.DataFrame"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "1190db37-b3e6-45a1-bbc0-96a5a4091bcc",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Date</th>\n",
       "      <th>DayOfWeek</th>\n",
       "      <th>DepTime</th>\n",
       "      <th>CRSDepTime</th>\n",
       "      <th>ArrTime</th>\n",
       "      <th>CRSArrTime</th>\n",
       "      <th>UniqueCarrier</th>\n",
       "      <th>FlightNum</th>\n",
       "      <th>TailNum</th>\n",
       "      <th>ActualElapsedTime</th>\n",
       "      <th>...</th>\n",
       "      <th>AirTime</th>\n",
       "      <th>ArrDelay</th>\n",
       "      <th>DepDelay</th>\n",
       "      <th>Origin</th>\n",
       "      <th>Dest</th>\n",
       "      <th>Distance</th>\n",
       "      <th>TaxiIn</th>\n",
       "      <th>TaxiOut</th>\n",
       "      <th>Cancelled</th>\n",
       "      <th>Diverted</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1990-01-01</td>\n",
       "      <td>1</td>\n",
       "      <td>1621.0</td>\n",
       "      <td>1540</td>\n",
       "      <td>1747.0</td>\n",
       "      <td>1701</td>\n",
       "      <td>US</td>\n",
       "      <td>33</td>\n",
       "      <td>NaN</td>\n",
       "      <td>86.0</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>46.0</td>\n",
       "      <td>41.0</td>\n",
       "      <td>EWR</td>\n",
       "      <td>PIT</td>\n",
       "      <td>319.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1990-01-02</td>\n",
       "      <td>2</td>\n",
       "      <td>1547.0</td>\n",
       "      <td>1540</td>\n",
       "      <td>1700.0</td>\n",
       "      <td>1701</td>\n",
       "      <td>US</td>\n",
       "      <td>33</td>\n",
       "      <td>NaN</td>\n",
       "      <td>73.0</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>-1.0</td>\n",
       "      <td>7.0</td>\n",
       "      <td>EWR</td>\n",
       "      <td>PIT</td>\n",
       "      <td>319.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1990-01-03</td>\n",
       "      <td>3</td>\n",
       "      <td>1546.0</td>\n",
       "      <td>1540</td>\n",
       "      <td>1710.0</td>\n",
       "      <td>1701</td>\n",
       "      <td>US</td>\n",
       "      <td>33</td>\n",
       "      <td>NaN</td>\n",
       "      <td>84.0</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>9.0</td>\n",
       "      <td>6.0</td>\n",
       "      <td>EWR</td>\n",
       "      <td>PIT</td>\n",
       "      <td>319.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1990-01-04</td>\n",
       "      <td>4</td>\n",
       "      <td>1542.0</td>\n",
       "      <td>1540</td>\n",
       "      <td>1710.0</td>\n",
       "      <td>1701</td>\n",
       "      <td>US</td>\n",
       "      <td>33</td>\n",
       "      <td>NaN</td>\n",
       "      <td>88.0</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>9.0</td>\n",
       "      <td>2.0</td>\n",
       "      <td>EWR</td>\n",
       "      <td>PIT</td>\n",
       "      <td>319.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>1990-01-05</td>\n",
       "      <td>5</td>\n",
       "      <td>1549.0</td>\n",
       "      <td>1540</td>\n",
       "      <td>1706.0</td>\n",
       "      <td>1701</td>\n",
       "      <td>US</td>\n",
       "      <td>33</td>\n",
       "      <td>NaN</td>\n",
       "      <td>77.0</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>5.0</td>\n",
       "      <td>9.0</td>\n",
       "      <td>EWR</td>\n",
       "      <td>PIT</td>\n",
       "      <td>319.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>1990-01-06</td>\n",
       "      <td>6</td>\n",
       "      <td>1539.0</td>\n",
       "      <td>1540</td>\n",
       "      <td>1653.0</td>\n",
       "      <td>1701</td>\n",
       "      <td>US</td>\n",
       "      <td>33</td>\n",
       "      <td>NaN</td>\n",
       "      <td>74.0</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>-8.0</td>\n",
       "      <td>-1.0</td>\n",
       "      <td>EWR</td>\n",
       "      <td>PIT</td>\n",
       "      <td>319.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>1990-01-07</td>\n",
       "      <td>7</td>\n",
       "      <td>1553.0</td>\n",
       "      <td>1540</td>\n",
       "      <td>1713.0</td>\n",
       "      <td>1701</td>\n",
       "      <td>US</td>\n",
       "      <td>33</td>\n",
       "      <td>NaN</td>\n",
       "      <td>80.0</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>12.0</td>\n",
       "      <td>13.0</td>\n",
       "      <td>EWR</td>\n",
       "      <td>PIT</td>\n",
       "      <td>319.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>1990-01-08</td>\n",
       "      <td>1</td>\n",
       "      <td>1543.0</td>\n",
       "      <td>1540</td>\n",
       "      <td>1656.0</td>\n",
       "      <td>1701</td>\n",
       "      <td>US</td>\n",
       "      <td>33</td>\n",
       "      <td>NaN</td>\n",
       "      <td>73.0</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>-5.0</td>\n",
       "      <td>3.0</td>\n",
       "      <td>EWR</td>\n",
       "      <td>PIT</td>\n",
       "      <td>319.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>1990-01-09</td>\n",
       "      <td>2</td>\n",
       "      <td>1540.0</td>\n",
       "      <td>1540</td>\n",
       "      <td>1704.0</td>\n",
       "      <td>1701</td>\n",
       "      <td>US</td>\n",
       "      <td>33</td>\n",
       "      <td>NaN</td>\n",
       "      <td>84.0</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>3.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>EWR</td>\n",
       "      <td>PIT</td>\n",
       "      <td>319.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>1990-01-10</td>\n",
       "      <td>3</td>\n",
       "      <td>1608.0</td>\n",
       "      <td>1540</td>\n",
       "      <td>1740.0</td>\n",
       "      <td>1701</td>\n",
       "      <td>US</td>\n",
       "      <td>33</td>\n",
       "      <td>NaN</td>\n",
       "      <td>92.0</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>39.0</td>\n",
       "      <td>28.0</td>\n",
       "      <td>EWR</td>\n",
       "      <td>PIT</td>\n",
       "      <td>319.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>10 rows × 21 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "        Date  DayOfWeek  DepTime  CRSDepTime  ArrTime  CRSArrTime  \\\n",
       "0 1990-01-01          1   1621.0        1540   1747.0        1701   \n",
       "1 1990-01-02          2   1547.0        1540   1700.0        1701   \n",
       "2 1990-01-03          3   1546.0        1540   1710.0        1701   \n",
       "3 1990-01-04          4   1542.0        1540   1710.0        1701   \n",
       "4 1990-01-05          5   1549.0        1540   1706.0        1701   \n",
       "5 1990-01-06          6   1539.0        1540   1653.0        1701   \n",
       "6 1990-01-07          7   1553.0        1540   1713.0        1701   \n",
       "7 1990-01-08          1   1543.0        1540   1656.0        1701   \n",
       "8 1990-01-09          2   1540.0        1540   1704.0        1701   \n",
       "9 1990-01-10          3   1608.0        1540   1740.0        1701   \n",
       "\n",
       "  UniqueCarrier  FlightNum TailNum  ActualElapsedTime  ...  AirTime  ArrDelay  \\\n",
       "0            US         33     NaN               86.0  ...      NaN      46.0   \n",
       "1            US         33     NaN               73.0  ...      NaN      -1.0   \n",
       "2            US         33     NaN               84.0  ...      NaN       9.0   \n",
       "3            US         33     NaN               88.0  ...      NaN       9.0   \n",
       "4            US         33     NaN               77.0  ...      NaN       5.0   \n",
       "5            US         33     NaN               74.0  ...      NaN      -8.0   \n",
       "6            US         33     NaN               80.0  ...      NaN      12.0   \n",
       "7            US         33     NaN               73.0  ...      NaN      -5.0   \n",
       "8            US         33     NaN               84.0  ...      NaN       3.0   \n",
       "9            US         33     NaN               92.0  ...      NaN      39.0   \n",
       "\n",
       "   DepDelay  Origin Dest Distance  TaxiIn  TaxiOut  Cancelled  Diverted  \n",
       "0      41.0     EWR  PIT    319.0     NaN      NaN      False         0  \n",
       "1       7.0     EWR  PIT    319.0     NaN      NaN      False         0  \n",
       "2       6.0     EWR  PIT    319.0     NaN      NaN      False         0  \n",
       "3       2.0     EWR  PIT    319.0     NaN      NaN      False         0  \n",
       "4       9.0     EWR  PIT    319.0     NaN      NaN      False         0  \n",
       "5      -1.0     EWR  PIT    319.0     NaN      NaN      False         0  \n",
       "6      13.0     EWR  PIT    319.0     NaN      NaN      False         0  \n",
       "7       3.0     EWR  PIT    319.0     NaN      NaN      False         0  \n",
       "8       0.0     EWR  PIT    319.0     NaN      NaN      False         0  \n",
       "9      28.0     EWR  PIT    319.0     NaN      NaN      False         0  \n",
       "\n",
       "[10 rows x 21 columns]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "result.compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "9720ec49-3997-4bec-98ac-e69345127996",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Origin</th>\n",
       "      <th>avg_dep_delay</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>EWR</td>\n",
       "      <td>10.295469</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>JFK</td>\n",
       "      <td>10.351299</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>LGA</td>\n",
       "      <td>7.431142</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  Origin  avg_dep_delay\n",
       "0    EWR      10.295469\n",
       "1    JFK      10.351299\n",
       "2    LGA       7.431142"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query = '''\n",
    "SELECT\n",
    "    Origin\n",
    "    , AVG(DepDelay) AS avg_dep_delay\n",
    "FROM\n",
    "    flights\n",
    "GROUP BY Origin\n",
    "'''\n",
    "\n",
    "result = c.sql(query)\n",
    "result.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f0f79185-0e39-4a0b-a639-1b1ca6941919",
   "metadata": {},
   "source": [
    "### What about \"creating the table completely in SQL\"?\n",
    "\n",
    "First, let's go \"full SQL\" so we don't even need to wrap our queries in Python..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "eb3f7a1f-c80a-480a-8c7b-7d4c475d72df",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/javascript": [
       "\n",
       "    require([\"codemirror/lib/codemirror\"]);\n",
       "\n",
       "    // We define a new mime type for syntax highlighting\n",
       "    CodeMirror.defineMIME(\"text/x-dasksql\", {\"name\": \"sql\", \"keywords\": {\"and\": true, \"as\": true, \"asc\": true, \"between\": true, \"by\": true, \"columns\": true, \"count\": true, \"create\": true, \"delete\": true, \"desc\": true, \"describe\": true, \"distinct\": true, \"exists\": true, \"from\": true, \"group\": true, \"having\": true, \"if\": true, \"in\": true, \"inner\": true, \"insert\": true, \"into\": true, \"is\": true, \"join\": true, \"left\": true, \"like\": true, \"model\": true, \"not\": true, \"on\": true, \"or\": true, \"order\": true, \"outer\": true, \"right\": true, \"schemas\": true, \"select\": true, \"set\": true, \"show\": true, \"table\": true, \"union\": true, \"where\": true, \">\": true, \">=\": true, \"<\": true, \"<=\": true, \"=\": true, \"<>\": true, \"+\": true, \"-\": true, \"/\": true, \"*\": true, \"is distinct from\": true, \"is not distinct from\": true, \"cast\": true, \"case\": true, \"similar to\": true, \"is null\": true, \"is not null\": true, \"is true\": true, \"is not true\": true, \"is false\": true, \"is not false\": true, \"is unknown\": true, \"is not unknown\": true, \"rand\": true, \"rand_integer\": true, \"abs\": true, \"acos\": true, \"asin\": true, \"atan\": true, \"atan2\": true, \"cbrt\": true, \"ceil\": true, \"cos\": true, \"cot\": true, \"degrees\": true, \"exp\": true, \"floor\": true, \"log10\": true, \"ln\": true, \"mod\": true, \"power\": true, \"radians\": true, \"round\": true, \"sign\": true, \"sin\": true, \"tan\": true, \"truncate\": true, \"||\": true, \"char_length\": true, \"upper\": true, \"lower\": true, \"position\": true, \"trim\": true, \"overlay\": true, \"substring\": true, \"initcap\": true, \"extract\": true, \"localtime\": true, \"localtimestamp\": true, \"current_time\": true, \"current_date\": true, \"current_timestamp\": true, \"last_day\": true}, \"builtin\": {\"DOUBLE\": true, \"FLOAT\": true, \"DECIMAL\": true, \"BIGINT\": true, \"INTEGER\": true, \"INT\": true, \"SMALLINT\": true, \"TINYINT\": true, \"BOOLEAN\": true, \"VARCHAR\": true, \"CHAR\": true, \"STRING\": true, \"DATE\": true, \"TIMESTAMP\": true, \"NULL\": true}, \"atoms\": {\"false\": true, \"true\": true, \"null\": true}, \"dateSQL\": {\"time\": true}, \"support\": {\"ODBCdotTable\": true, \"doubleQuote\": true, \"zerolessFloat\": true}}\n",
       "    );\n",
       "    CodeMirror.modeInfo.push({\n",
       "        name: \"Dask SQL\",\n",
       "        mime: \"text/x-dasksql\",\n",
       "        mode: \"sql\"\n",
       "    });\n",
       "    \n",
       "require(['notebook/js/codecell'], function(codecell) {\n",
       "    codecell.CodeCell.options_default.highlight_modes['magic_text/x-dasksql'] = {'reg':[/%%sql/]} ;\n",
       "    Jupyter.notebook.events.on('kernel_ready.Kernel', function(){\n",
       "    Jupyter.notebook.get_cells().map(function(cell){\n",
       "        if (cell.cell_type == 'code'){ cell.auto_highlight(); } }) ;\n",
       "    });\n",
       "});\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "c.ipython_magic()  # ignore the error?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "fe94eb3e-4309-42cd-9ce8-166d750a6c65",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "\n",
    "CREATE TABLE allsql WITH (\n",
    "    format = 'csv',\n",
    "    location = 'data/nycflights/1990.csv' -- any Dask-accessible source or format (cloud/S3/..., parquet/ORC/...)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "79f0aed9-d3bb-4d84-bed9-2bafd66bb04d",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Origin</th>\n",
       "      <th>avg_dep_delay</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>EWR</td>\n",
       "      <td>9.168411</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>JFK</td>\n",
       "      <td>11.857274</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>LGA</td>\n",
       "      <td>8.560045</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  Origin  avg_dep_delay\n",
       "0    EWR       9.168411\n",
       "1    JFK      11.857274\n",
       "2    LGA       8.560045"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%sql\n",
    "\n",
    "SELECT\n",
    "    Origin\n",
    "    , AVG(DepDelay) AS avg_dep_delay\n",
    "FROM\n",
    "    allsql\n",
    "GROUP BY Origin"
   ]
  }
 ],
 "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.9.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}