notebooks/Feature Overview.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# dask-sql\n",
    "### A SQL Query Layer for Dask"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Introduction"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`dask-sql` adds a SQL query layer on top of the Dask distributed Python library, which allows you to query your big and small data with SQL and still use the great power of the Dask ecosystem.\n",
    "It helps you combine the best of both worlds.\n",
    "See the [documentation](https://dask-sql.readthedocs.io/) for more information."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Starting dask-sql"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There are two possibilities how you can send your SQL queries to `dask-sql`:\n",
    "* you use a Python notebook/script, such as the one you have currently opened\n",
    "* you run the [dask-sql Server](https://dask-sql.readthedocs.io/en/latest/pages/server.html) as a standalone application and connect to it via e.g. your BI tool\n",
    "\n",
    "We will stick with the first possibility in this notebook, but all SQL commands shown here can also be run via the SQL server."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Before we start, we need do import `dask-sql` and create a `Context`, which collects all the information on the currently registered data tables.\n",
    "We will also create a small local Dask cluster (this step is not needed, but gives us a bit more debugging options).\n",
    "If you have a large computation cluster, you can connect to it in this step (have a look [here](https://docs.dask.org/en/latest/setup.html))."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask_sql import Context\n",
    "from dask.distributed import Client\n",
    "\n",
    "client = Client()\n",
    "c = Context()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You are now ready to query with SQL!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c.sql(\"\"\"\n",
    "    SELECT 42 AS \"the answer\"\n",
    "\"\"\", return_futures=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Some shortcut for the following:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c.ipython_magic(auto_include=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This line allows us to write (instead of the line above)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "SELECT 42 AS \"the answer\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Input"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1. From a Dask Dataframe via Python"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.dataframe as dd\n",
    "\n",
    "df = dd.read_csv(\"./iris.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.head(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c.create_table(\"iris\", df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. From an external data source via SQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "CREATE OR REPLACE TABLE iris\n",
    "WITH (\n",
    "    location = 'file://./iris.csv',\n",
    "    format = 'csv',\n",
    "    persist = True\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* s3, azure, dbfs (new!), gs, hdfs, ...\n",
    "* hive (experimental), databricks (experimental), intake\n",
    "* already loaded data persisted in your Dask cluster\n",
    "\n",
    "More [information](https://dask-sql.readthedocs.io/en/latest/pages/data_input.html)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3. As materialized Queries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "CREATE OR REPLACE TABLE second_iris\n",
    "AS SELECT * FROM iris"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4. From the notebook"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we have created an ipython magic with `c.ipython_magic(auto_include=True)` we can even just reference any dataframe created in the notebook in our queries."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "my_data_frame = dd.read_csv(\"./iris.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "SELECT * FROM my_data_frame LIMIT 10"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Please note that using this setting will automatically override any predefined tables with the same name."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Metadata Information"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "SHOW TABLES FROM \"schema\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "SHOW COLUMNS FROM \"iris\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "DESCRIBE iris"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "DESCRIBE TABLE iris"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Query"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can call \"normal\" SQL `SELECT` statements in `dask-sql`, with all typical components from the standard SQL language.\n",
    "More information in the [SQL reference](https://dask-sql.readthedocs.io/en/latest/pages/sql.html).\n",
    "`dask-sql` roughly follows the prestoSQL conventions (e.g. quoting).\n",
    "\n",
    "<div class=\"alert alert-info\">\n",
    "    \n",
    "#### Note\n",
    "    \n",
    "Not all SQL operators are implemented in `dask-sql` already.\n",
    "    \n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "SELECT * \n",
    "FROM iris\n",
    "LIMIT 10"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "SELECT \n",
    "    sepal_length + sepal_width AS \"sum\", \n",
    "    SIN(petal_length) AS \"sin\"\n",
    "FROM iris\n",
    "LIMIT 10"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "SELECT \n",
    "    species,\n",
    "    AVG(sepal_length) AS sepal_length, \n",
    "    AVG(sepal_width) AS sepal_width\n",
    "FROM iris\n",
    "GROUP BY species\n",
    "LIMIT 10"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "WITH maximal_values AS (\n",
    "    SELECT \n",
    "        species, \n",
    "        MAX(sepal_length) AS sepal_length\n",
    "    FROM iris\n",
    "    GROUP BY species\n",
    ")\n",
    "SELECT lhs.*\n",
    "FROM iris AS lhs \n",
    "JOIN maximal_values AS rhs ON lhs.species = rhs.species AND lhs.sepal_length = rhs.sepal_length"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(c.explain(\"\"\"\n",
    "    WITH maximal_values AS (\n",
    "        SELECT \n",
    "            species, \n",
    "            MAX(sepal_length) AS sepal_length\n",
    "        FROM iris\n",
    "        GROUP BY species\n",
    "    )\n",
    "    SELECT \n",
    "        lhs.*\n",
    "    FROM iris AS lhs \n",
    "    JOIN maximal_values AS rhs\n",
    "    ON lhs.species = rhs.species \n",
    "        AND lhs.sepal_length = rhs.sepal_length\n",
    "\"\"\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Custom Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "\n",
    "def volume(length, width):\n",
    "    return (width / 2) ** 2 * np.pi * length\n",
    "\n",
    "# As SQL is a typed language, we need to specify all types \n",
    "c.register_function(volume, \"IRIS_VOLUME\", \n",
    "                    parameters=[(\"length\", np.float64), (\"width\", np.float64)], \n",
    "                    return_type=np.float64)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "SELECT \n",
    "    sepal_length, sepal_width, IRIS_VOLUME(sepal_length, sepal_width) AS volume\n",
    "FROM iris\n",
    "LIMIT 10"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Machine Learning"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.species.head(100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "CREATE OR REPLACE TABLE enriched_iris AS (\n",
    "    SELECT \n",
    "        sepal_length, sepal_width, petal_length, petal_width,\n",
    "        CASE \n",
    "            WHEN species = 'setosa' THEN 0 ELSE CASE \n",
    "            WHEN species = 'versicolor' THEN 1\n",
    "            ELSE 2 \n",
    "        END END AS \"species\", \n",
    "        IRIS_VOLUME(sepal_length, sepal_width) AS volume\n",
    "    FROM iris \n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "CREATE OR REPLACE TABLE training_data AS (\n",
    "    SELECT \n",
    "        *\n",
    "    FROM enriched_iris\n",
    "    TABLESAMPLE BERNOULLI (50)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "SELECT * FROM training_data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "CREATE OR REPLACE MODEL my_model WITH (\n",
    "    model_class = 'xgboost.dask.DaskXGBClassifier',\n",
    "    target_column = 'species',\n",
    "    num_class = 3\n",
    ") AS (\n",
    "    SELECT * FROM training_data\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "SHOW MODELS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "DESCRIBE MODEL my_model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "SELECT\n",
    "    *\n",
    "FROM PREDICT(\n",
    "    MODEL my_model,\n",
    "    SELECT * FROM enriched_iris\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "CREATE OR REPLACE TABLE results AS\n",
    "SELECT\n",
    "    *\n",
    "FROM PREDICT(\n",
    "    MODEL my_model,\n",
    "    TABLE enriched_iris\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create Experiment "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Tune single model with different Hyperparameters \n",
    "  - install **sklearn** for tuning\n",
    "- Tune multiple model with different Hyperparameters\n",
    "  - install **tpot** for Automl"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "CREATE EXPERIMENT my_exp WITH (\n",
    "        model_class = 'sklearn.ensemble.GradientBoostingClassifier',\n",
    "        experiment_class = 'sklearn.model_selection.GridSearchCV',\n",
    "        tune_parameters = (n_estimators = ARRAY [16, 32, 2],learning_rate = ARRAY [0.1,0.01,0.001],\n",
    "                           max_depth = ARRAY [3,4,5,10]),\n",
    "        target_column = 'species'\n",
    "    ) AS (\n",
    "            SELECT * FROM training_data\n",
    "        )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Automl in SQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "CREATE EXPERIMENT my_automl_exp WITH (\n",
    "            automl_class = 'tpot.TPOTClassifier',\n",
    "            automl_kwargs = (population_size = 2 ,generations=5,cv=2,n_jobs=-1,use_dask=True),\n",
    "            target_column = 'species'\n",
    "        ) AS (\n",
    "            SELECT * FROM training_data \n",
    "        )\n",
    "-- while model was training, checkout the visualization in dask-dashboard for each generation of tasks "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql \n",
    "show models\n",
    "\n",
    "-- once the experiment was completed, Best model was\n",
    "-- stored in context which can be used for prediction"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Export ML models\n",
    "\n",
    "- export trained models and serve the model as microservice\n",
    "- supports pickle,joblib,mlflow formats"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "-- pickle  export\n",
    "EXPORT MODEL my_model with (\n",
    "            format ='pickle',\n",
    "            location = 'my_model.pkl'\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "-- joblib export\n",
    "EXPORT MODEL my_exp_GradientBoostingClassifier_best_model\n",
    "with (\n",
    "            format ='joblib',\n",
    "            location = 'best_mode.joblib'\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "-- mlflow export\n",
    "EXPORT MODEL automl_TPOTClassifier\n",
    "with (\n",
    "            format ='mlflow',\n",
    "            location = 'model_dir'\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "SELECT\n",
    "    target,\n",
    "    species,\n",
    "    COUNT(*)\n",
    "FROM\n",
    "    results\n",
    "GROUP BY target, species"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t = c.sql(\"\"\"\n",
    "    SELECT\n",
    "        target,\n",
    "        species,\n",
    "        COUNT(*) AS \"number\"\n",
    "    FROM\n",
    "        results\n",
    "    GROUP BY target, species\n",
    "\"\"\").compute() \n",
    "t.set_index([\"target\", \"species\"]).unstack(\"species\").number.plot.bar()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "Python 3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.6"
  },
  "toc-autonumbering": false,
  "toc-showcode": false,
  "toc-showmarkdowntxt": false
 },
 "nbformat": 4,
 "nbformat_minor": 4
}