examples/wine-ratings-prediction/Predict Missing Wine Ratings.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "9ad8a3e7-af62-4a4d-a05e-f5ca5fc477a5",
   "metadata": {},
   "source": [
    "# Predict Missing Wine Ratings\n",
    "\n",
    "In this demo, we train and use multiple models to impute missing values.  We start with a dataset of wines consisting of key features like acidity. Some of the records are missing feature values. In addition, wine quality scores are given to some but not all of the wines. \n",
    "\n",
    "**You can find and download this notebook on GitHub [here](https://github.com/aqueducthq/aqueduct/blob/main/examples/training_and_inference/Impute%20Missing%20Wine%20Data.ipynb).**\n",
    "\n",
    "We will build a workflow that trains a linear model to impute the missing features from the other features and then train a decision tree model to rate the un-rated wines using the imputed features. \n",
    "\n",
    "**Throughout this notebook, you'll see a decorator (`@aq.op`) above functions. This decorator allows Aqueduct to run your functions as a part of a workflow automatically.**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "6659a9ca-cfe7-4cdc-ac05-563a4fe851e3",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import aqueduct\n",
    "from aqueduct import op, check, metric"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "0baae647-283e-4ce7-b554-2939b262173f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# You can use `localhost` if you're running this notebook on the same machine as the server.\n",
    "# If you're running your notebook on a separate machine from your\n",
    "# Aqueduct server, change this to the address of your Aqueduct server.\n",
    "address = \"http://localhost:8080\"\n",
    "\n",
    "# If you're running youre notebook on a separate machine from your\n",
    "# Aqueduct server, you will have to copy your API key here rather than\n",
    "# using `get_apikey()`.\n",
    "api_key = aqueduct.get_apikey()\n",
    "client = aqueduct.Client(api_key, address)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b28d68b1-5efb-4166-b6b5-ad11cc55aa60",
   "metadata": {},
   "source": [
    "## Getting the Demo Data \n",
    "\n",
    "In this demo, we will use the wine table in the demo data warehouse."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "d5ed9715-8009-4318-892d-9dfb63f8b1d4",
   "metadata": {},
   "outputs": [],
   "source": [
    "demodb = client.resource(\"Demo\")\n",
    "\n",
    "# wines is an Aqueduct TableArtifact, which is a wrapper around\n",
    "# a Pandas DataFrame. A TableArtifact can be used as argument to any operator\n",
    "# in a workflow; you can also call .get() on a TableArtifact to retrieve\n",
    "# the underlying DataFrame and interact with it directly.\n",
    "wines = demodb.sql(\"select * from wine;\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "b784eb0f-1f41-4851-ad98-e9448e540de8",
   "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>id</th>\n",
       "      <th>color</th>\n",
       "      <th>fixed_acidity</th>\n",
       "      <th>volatile_acidity</th>\n",
       "      <th>citric_acid</th>\n",
       "      <th>residual_sugar</th>\n",
       "      <th>chlorides</th>\n",
       "      <th>free_sulfur_dioxide</th>\n",
       "      <th>total_sulfur_dioxide</th>\n",
       "      <th>density</th>\n",
       "      <th>ph</th>\n",
       "      <th>sulphates</th>\n",
       "      <th>alcohol</th>\n",
       "      <th>quality</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>white</td>\n",
       "      <td>7.0</td>\n",
       "      <td>0.17</td>\n",
       "      <td>0.74</td>\n",
       "      <td>\\N</td>\n",
       "      <td>0.045</td>\n",
       "      <td>24.0</td>\n",
       "      <td>126.0</td>\n",
       "      <td>0.99420</td>\n",
       "      <td>3.26</td>\n",
       "      <td>0.38</td>\n",
       "      <td>12.2</td>\n",
       "      <td>8.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>red</td>\n",
       "      <td>7.7</td>\n",
       "      <td>0.64</td>\n",
       "      <td>0.21</td>\n",
       "      <td>2.2</td>\n",
       "      <td>0.077</td>\n",
       "      <td>32.0</td>\n",
       "      <td>133.0</td>\n",
       "      <td>0.99560</td>\n",
       "      <td>3.27</td>\n",
       "      <td>0.45</td>\n",
       "      <td>9.9</td>\n",
       "      <td>5.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2</td>\n",
       "      <td>white</td>\n",
       "      <td>6.8</td>\n",
       "      <td>0.39</td>\n",
       "      <td>0.34</td>\n",
       "      <td>7.4</td>\n",
       "      <td>0.020</td>\n",
       "      <td>38.0</td>\n",
       "      <td>133.0</td>\n",
       "      <td>0.99212</td>\n",
       "      <td>3.18</td>\n",
       "      <td>0.44</td>\n",
       "      <td>12.0</td>\n",
       "      <td>\\N</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>3</td>\n",
       "      <td>white</td>\n",
       "      <td>6.3</td>\n",
       "      <td>0.28</td>\n",
       "      <td>0.47</td>\n",
       "      <td>\\N</td>\n",
       "      <td>0.040</td>\n",
       "      <td>61.0</td>\n",
       "      <td>183.0</td>\n",
       "      <td>0.99592</td>\n",
       "      <td>3.12</td>\n",
       "      <td>0.51</td>\n",
       "      <td>9.5</td>\n",
       "      <td>6.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>4</td>\n",
       "      <td>white</td>\n",
       "      <td>7.4</td>\n",
       "      <td>0.35</td>\n",
       "      <td>0.20</td>\n",
       "      <td>13.9</td>\n",
       "      <td>0.054</td>\n",
       "      <td>63.0</td>\n",
       "      <td>229.0</td>\n",
       "      <td>0.99888</td>\n",
       "      <td>3.11</td>\n",
       "      <td>0.50</td>\n",
       "      <td>8.9</td>\n",
       "      <td>\\N</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   id  color  fixed_acidity  volatile_acidity  citric_acid residual_sugar  \\\n",
       "0   0  white            7.0              0.17         0.74             \\N   \n",
       "1   1    red            7.7              0.64         0.21            2.2   \n",
       "2   2  white            6.8              0.39         0.34            7.4   \n",
       "3   3  white            6.3              0.28         0.47             \\N   \n",
       "4   4  white            7.4              0.35         0.20           13.9   \n",
       "\n",
       "   chlorides  free_sulfur_dioxide  total_sulfur_dioxide  density    ph  \\\n",
       "0      0.045                 24.0                 126.0  0.99420  3.26   \n",
       "1      0.077                 32.0                 133.0  0.99560  3.27   \n",
       "2      0.020                 38.0                 133.0  0.99212  3.18   \n",
       "3      0.040                 61.0                 183.0  0.99592  3.12   \n",
       "4      0.054                 63.0                 229.0  0.99888  3.11   \n",
       "\n",
       "   sulphates  alcohol quality  \n",
       "0       0.38     12.2     8.0  \n",
       "1       0.45      9.9     5.0  \n",
       "2       0.44     12.0      \\N  \n",
       "3       0.51      9.5     6.0  \n",
       "4       0.50      8.9      \\N  "
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# This gets the head of the underlying DataFrame. Note that you can't\n",
    "# pass a DataFrame as an argument to a workflow; you must use the Aqueduct\n",
    "# TableArtifact!\n",
    "wines.get().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "447fe1e4-434e-4a57-a7af-005cc99ebdf8",
   "metadata": {},
   "source": [
    "Notice that we are missing some values in the `residual_sugar` column as well as the `quality` column which we would ultimately like to predict."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e69a20bc-2246-4eb9-b94a-21805c324c92",
   "metadata": {},
   "source": [
    "## Cleaning the Data\n",
    "There are some missing values in the residula sugar column that we need to clean.  Here we will replace the residual sugar with a value predicted by other columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "5ee0f6e9-c379-47c3-8a04-667d78e8cd02",
   "metadata": {},
   "outputs": [],
   "source": [
    "# The @op decorator here allows Aqueduct to run this function as\n",
    "# a part of the Aqueduct workflow. It tells Aqueduct that when\n",
    "# we execute this function, we're defining a step in the workflow.\n",
    "# While the results can be retrieved immediately, nothing is\n",
    "# published until we call `publish_flow()` below.\n",
    "@op()\n",
    "def fix_residual_sugar(df):\n",
    "    \"\"\"\n",
    "    This function takes in a DataFrame representing wines data and cleans\n",
    "    the DataFrame by replacing any missing values in the `residual_sugar`\n",
    "    column with the values that would be predicted based on the other columns.\n",
    "\n",
    "    Internally, this function uses the sklearn LinearRegression model to\n",
    "    predict what the values of the `residual_sugar` column should be when\n",
    "    they are missing.\n",
    "    \"\"\"\n",
    "    from sklearn.linear_model import LinearRegression\n",
    "\n",
    "    # Convert residual_sugar back to numeric values with missing values as NaN\n",
    "    df[\"residual_sugar\"] = pd.to_numeric(df[\"residual_sugar\"], errors=\"coerce\")\n",
    "    print(\"missing residual sugar values:\", df[\"residual_sugar\"].isna().sum())\n",
    "\n",
    "    # Fit a LinearRegression model on the other numeric columns, which is everything but\n",
    "    # quality, residual_sugar.\n",
    "    imputer = LinearRegression()\n",
    "    other_cols = df.columns[df.dtypes == \"float\"].difference([\"quality\", \"residual_sugar\", \"id\"])\n",
    "    imputer.fit(df.dropna()[other_cols], df.dropna()[\"residual_sugar\"])\n",
    "\n",
    "    # Use our newly-trained imputer to predict the missing values of `residual_sugar`\n",
    "    # and replace the NaN values with our new predicted values.\n",
    "    predicted_sugar = imputer.predict(df[df[\"residual_sugar\"].isna()][other_cols])\n",
    "    df.loc[df[\"residual_sugar\"].isna(), \"residual_sugar\"] = predicted_sugar\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "56695cf6-cc3a-4725-aab8-886adfd9316a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Operator fix_residual_sugar Logs:\n",
      "stdout:\n",
      "            missing residual sugar values: 500\n",
      "\n",
      "            --------------------------\n",
      "            stderr:\n",
      "\n",
      "\n"
     ]
    },
    {
     "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>id</th>\n",
       "      <th>color</th>\n",
       "      <th>fixed_acidity</th>\n",
       "      <th>volatile_acidity</th>\n",
       "      <th>citric_acid</th>\n",
       "      <th>residual_sugar</th>\n",
       "      <th>chlorides</th>\n",
       "      <th>free_sulfur_dioxide</th>\n",
       "      <th>total_sulfur_dioxide</th>\n",
       "      <th>density</th>\n",
       "      <th>ph</th>\n",
       "      <th>sulphates</th>\n",
       "      <th>alcohol</th>\n",
       "      <th>quality</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>white</td>\n",
       "      <td>7.0</td>\n",
       "      <td>0.17</td>\n",
       "      <td>0.74</td>\n",
       "      <td>10.027690</td>\n",
       "      <td>0.045</td>\n",
       "      <td>24.0</td>\n",
       "      <td>126.0</td>\n",
       "      <td>0.99420</td>\n",
       "      <td>3.26</td>\n",
       "      <td>0.38</td>\n",
       "      <td>12.2</td>\n",
       "      <td>8.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>red</td>\n",
       "      <td>7.7</td>\n",
       "      <td>0.64</td>\n",
       "      <td>0.21</td>\n",
       "      <td>2.200000</td>\n",
       "      <td>0.077</td>\n",
       "      <td>32.0</td>\n",
       "      <td>133.0</td>\n",
       "      <td>0.99560</td>\n",
       "      <td>3.27</td>\n",
       "      <td>0.45</td>\n",
       "      <td>9.9</td>\n",
       "      <td>5.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2</td>\n",
       "      <td>white</td>\n",
       "      <td>6.8</td>\n",
       "      <td>0.39</td>\n",
       "      <td>0.34</td>\n",
       "      <td>7.400000</td>\n",
       "      <td>0.020</td>\n",
       "      <td>38.0</td>\n",
       "      <td>133.0</td>\n",
       "      <td>0.99212</td>\n",
       "      <td>3.18</td>\n",
       "      <td>0.44</td>\n",
       "      <td>12.0</td>\n",
       "      <td>\\N</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>3</td>\n",
       "      <td>white</td>\n",
       "      <td>6.3</td>\n",
       "      <td>0.28</td>\n",
       "      <td>0.47</td>\n",
       "      <td>10.697218</td>\n",
       "      <td>0.040</td>\n",
       "      <td>61.0</td>\n",
       "      <td>183.0</td>\n",
       "      <td>0.99592</td>\n",
       "      <td>3.12</td>\n",
       "      <td>0.51</td>\n",
       "      <td>9.5</td>\n",
       "      <td>6.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>4</td>\n",
       "      <td>white</td>\n",
       "      <td>7.4</td>\n",
       "      <td>0.35</td>\n",
       "      <td>0.20</td>\n",
       "      <td>13.900000</td>\n",
       "      <td>0.054</td>\n",
       "      <td>63.0</td>\n",
       "      <td>229.0</td>\n",
       "      <td>0.99888</td>\n",
       "      <td>3.11</td>\n",
       "      <td>0.50</td>\n",
       "      <td>8.9</td>\n",
       "      <td>\\N</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   id  color  fixed_acidity  volatile_acidity  citric_acid  residual_sugar  \\\n",
       "0   0  white            7.0              0.17         0.74       10.027690   \n",
       "1   1    red            7.7              0.64         0.21        2.200000   \n",
       "2   2  white            6.8              0.39         0.34        7.400000   \n",
       "3   3  white            6.3              0.28         0.47       10.697218   \n",
       "4   4  white            7.4              0.35         0.20       13.900000   \n",
       "\n",
       "   chlorides  free_sulfur_dioxide  total_sulfur_dioxide  density    ph  \\\n",
       "0      0.045                 24.0                 126.0  0.99420  3.26   \n",
       "1      0.077                 32.0                 133.0  0.99560  3.27   \n",
       "2      0.020                 38.0                 133.0  0.99212  3.18   \n",
       "3      0.040                 61.0                 183.0  0.99592  3.12   \n",
       "4      0.054                 63.0                 229.0  0.99888  3.11   \n",
       "\n",
       "   sulphates  alcohol quality  \n",
       "0       0.38     12.2     8.0  \n",
       "1       0.45      9.9     5.0  \n",
       "2       0.44     12.0      \\N  \n",
       "3       0.51      9.5     6.0  \n",
       "4       0.50      8.9      \\N  "
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# This tells Aqueduct to execute fix_residual_sugar on wines\n",
    "# as a part of our workflow. However, nothing is published (yet) until we\n",
    "# call `publish_flow()` below. Calling `.get()` gives us a preview\n",
    "# of the results, but the resulting value is not stored or published\n",
    "# anywhere.\n",
    "wines_cleaned = fix_residual_sugar(wines)\n",
    "wines_cleaned.get().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3d285ba4-8cf7-4849-99f9-42c23a44425a",
   "metadata": {},
   "source": [
    "We also want to encode the color column as a boolean rather than a string. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "f4f3bd9e-d460-4f00-be99-a126e300de0c",
   "metadata": {},
   "outputs": [],
   "source": [
    "@op()\n",
    "def encode_color(df):\n",
    "    \"\"\"\n",
    "    This function takes in a DataFrame with data about wines\n",
    "    and encodes whether the wine is a red wine or white wine\n",
    "    as boolean value. This allows us to treat this as a categorical\n",
    "    variable for a future model-training step.\n",
    "    \"\"\"\n",
    "    df[\"is_red\"] = (df[\"color\"] == \"red\").astype(\"float\")\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "d5fdf671-f08d-4e27-8ad6-65816d2d43cd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Operator fix_residual_sugar Logs:\n",
      "stdout:\n",
      "            missing residual sugar values: 500\n",
      "\n",
      "            --------------------------\n",
      "            stderr:\n",
      "\n",
      "\n"
     ]
    },
    {
     "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>id</th>\n",
       "      <th>color</th>\n",
       "      <th>fixed_acidity</th>\n",
       "      <th>volatile_acidity</th>\n",
       "      <th>citric_acid</th>\n",
       "      <th>residual_sugar</th>\n",
       "      <th>chlorides</th>\n",
       "      <th>free_sulfur_dioxide</th>\n",
       "      <th>total_sulfur_dioxide</th>\n",
       "      <th>density</th>\n",
       "      <th>ph</th>\n",
       "      <th>sulphates</th>\n",
       "      <th>alcohol</th>\n",
       "      <th>quality</th>\n",
       "      <th>is_red</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>white</td>\n",
       "      <td>7.0</td>\n",
       "      <td>0.17</td>\n",
       "      <td>0.74</td>\n",
       "      <td>10.027690</td>\n",
       "      <td>0.045</td>\n",
       "      <td>24.0</td>\n",
       "      <td>126.0</td>\n",
       "      <td>0.99420</td>\n",
       "      <td>3.26</td>\n",
       "      <td>0.38</td>\n",
       "      <td>12.2</td>\n",
       "      <td>8.0</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>red</td>\n",
       "      <td>7.7</td>\n",
       "      <td>0.64</td>\n",
       "      <td>0.21</td>\n",
       "      <td>2.200000</td>\n",
       "      <td>0.077</td>\n",
       "      <td>32.0</td>\n",
       "      <td>133.0</td>\n",
       "      <td>0.99560</td>\n",
       "      <td>3.27</td>\n",
       "      <td>0.45</td>\n",
       "      <td>9.9</td>\n",
       "      <td>5.0</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2</td>\n",
       "      <td>white</td>\n",
       "      <td>6.8</td>\n",
       "      <td>0.39</td>\n",
       "      <td>0.34</td>\n",
       "      <td>7.400000</td>\n",
       "      <td>0.020</td>\n",
       "      <td>38.0</td>\n",
       "      <td>133.0</td>\n",
       "      <td>0.99212</td>\n",
       "      <td>3.18</td>\n",
       "      <td>0.44</td>\n",
       "      <td>12.0</td>\n",
       "      <td>\\N</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>3</td>\n",
       "      <td>white</td>\n",
       "      <td>6.3</td>\n",
       "      <td>0.28</td>\n",
       "      <td>0.47</td>\n",
       "      <td>10.697218</td>\n",
       "      <td>0.040</td>\n",
       "      <td>61.0</td>\n",
       "      <td>183.0</td>\n",
       "      <td>0.99592</td>\n",
       "      <td>3.12</td>\n",
       "      <td>0.51</td>\n",
       "      <td>9.5</td>\n",
       "      <td>6.0</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>4</td>\n",
       "      <td>white</td>\n",
       "      <td>7.4</td>\n",
       "      <td>0.35</td>\n",
       "      <td>0.20</td>\n",
       "      <td>13.900000</td>\n",
       "      <td>0.054</td>\n",
       "      <td>63.0</td>\n",
       "      <td>229.0</td>\n",
       "      <td>0.99888</td>\n",
       "      <td>3.11</td>\n",
       "      <td>0.50</td>\n",
       "      <td>8.9</td>\n",
       "      <td>\\N</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   id  color  fixed_acidity  volatile_acidity  citric_acid  residual_sugar  \\\n",
       "0   0  white            7.0              0.17         0.74       10.027690   \n",
       "1   1    red            7.7              0.64         0.21        2.200000   \n",
       "2   2  white            6.8              0.39         0.34        7.400000   \n",
       "3   3  white            6.3              0.28         0.47       10.697218   \n",
       "4   4  white            7.4              0.35         0.20       13.900000   \n",
       "\n",
       "   chlorides  free_sulfur_dioxide  total_sulfur_dioxide  density    ph  \\\n",
       "0      0.045                 24.0                 126.0  0.99420  3.26   \n",
       "1      0.077                 32.0                 133.0  0.99560  3.27   \n",
       "2      0.020                 38.0                 133.0  0.99212  3.18   \n",
       "3      0.040                 61.0                 183.0  0.99592  3.12   \n",
       "4      0.054                 63.0                 229.0  0.99888  3.11   \n",
       "\n",
       "   sulphates  alcohol quality  is_red  \n",
       "0       0.38     12.2     8.0     0.0  \n",
       "1       0.45      9.9     5.0     1.0  \n",
       "2       0.44     12.0      \\N     0.0  \n",
       "3       0.51      9.5     6.0     0.0  \n",
       "4       0.50      8.9      \\N     0.0  "
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Again, we execute `encode_color` on `wines_cleaned` and use\n",
    "# `.get()` to retrieve a preview of the results, but no resulting\n",
    "# data is stored here.\n",
    "featurized_wines = encode_color(wines_cleaned)\n",
    "featurized_wines.get().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "97c47c44-4627-4278-bd6a-e1368453d0e5",
   "metadata": {},
   "source": [
    "## Tracking number of a Labeled wines \n",
    "\n",
    "As a sanity check, we want to make sure there are enough wines with quality scores to render reliable predictions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "2fee826a-9ea0-4677-8a1e-2f280a4b33af",
   "metadata": {},
   "outputs": [],
   "source": [
    "# The @metric dectorator is similar to the @op decorator from\n",
    "# above. The only difference is that a metric returns a numerical\n",
    "# value that is tracked over and visualized over time.\n",
    "@metric()\n",
    "def get_number_labeled_wines(df):\n",
    "    \"\"\"\n",
    "    This function takes in a DataFrame of wine data and returns\n",
    "    how many wines are missing a quality value. This function is based\n",
    "    on the assumption that missing values are encoded as `\\\\N` in the\n",
    "    underlying DataFrame. The typical, non-null value is expected to\n",
    "    be numeric.\n",
    "    \"\"\"\n",
    "    return (df[\"quality\"] != \"\\\\N\").sum().astype(float)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "42b62d7a-9f0f-478e-affe-77c46cf65f0f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Operator fix_residual_sugar Logs:\n",
      "stdout:\n",
      "            missing residual sugar values: 500\n",
      "\n",
      "            --------------------------\n",
      "            stderr:\n",
      "\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "5997.0"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Similar to executing an @op-annotated function, executing a\n",
    "# @metric-annotated function tells Aqueduct that get_number_labeled_wines\n",
    "# will be a part of our workflow but does not publish any information\n",
    "# till `publish_flow()` is called.\n",
    "num_labeled = get_number_labeled_wines(featurized_wines)\n",
    "num_labeled.get()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f4d8eaf7-6022-4cf3-92fe-195577fa93aa",
   "metadata": {},
   "source": [
    "To avoid making potentially risky predictions, we also don't want to render predictions if we have fewer than 1000 labeled wines. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "2d4b9848-4242-48d6-a310-f53455c93ec1",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<aqueduct.check_artifact.CheckArtifact at 0x1345126a0>"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# We add a lower bound of 1000 on our number of labeled wines.\n",
    "# If this value drops below 1000, the severity argument tells\n",
    "# us that we should automatically fail this workflow and not\n",
    "# publish the results.\n",
    "num_labeled.bound(lower=1000, severity=\"error\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "029bad8a-1163-44a0-8d6c-927ef6651447",
   "metadata": {},
   "source": [
    "## Predicting the Quality of Wines\n",
    "\n",
    "In the following operator we:\n",
    "1. Fit a decision tree model to the wines that do have quality ratings\n",
    "2. Make quality rating predictions for all the wines in the table."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "6b5ea39e-c80c-43c5-9562-12caa5c86056",
   "metadata": {},
   "outputs": [],
   "source": [
    "@op()\n",
    "def predict_quality(df):\n",
    "    \"\"\"\n",
    "    This function takes in data about wines and fills in any missing\n",
    "    values for the wine quality by building a machine learning model\n",
    "    that predicts the quality of the wine itself. The expectation for\n",
    "    this function is that many or most of the wines will already be labeled\n",
    "    with their quality. This function uses the existing wine quality\n",
    "    labels as guidance to train its model and fills in missing\n",
    "    values with the model.\n",
    "\n",
    "    Under the hood, this function uses sklearn's DecisionTreeRegressor\n",
    "    model to predict the missing wines' qualities.\n",
    "    \"\"\"\n",
    "    from sklearn.tree import DecisionTreeRegressor\n",
    "\n",
    "    # Convert the quality column to numerica and replace the \"\\N\" with NaN\n",
    "    df[\"quality\"] = pd.to_numeric(df[\"quality\"], errors=\"coerce\")\n",
    "\n",
    "    # Fit a model to the columns that are of numerical types but aren't the wine's\n",
    "    # ID or the quality that we're predicting.\n",
    "    quality_model = DecisionTreeRegressor(max_depth=3)\n",
    "    feature_columns = df.columns[df.dtypes == \"float\"].difference([\"quality\", \"id\"])\n",
    "    print(\"Feature Columns\", feature_columns)\n",
    "    quality_model.fit(df.dropna()[feature_columns], df.dropna()[\"quality\"])\n",
    "\n",
    "    # Add a `pred_quality` column with the predicted quality for each wine.\n",
    "    df[\"pred_quality\"] = quality_model.predict(df[feature_columns])\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "3abd3db2-e395-4bc0-b0b9-1119529e8ed5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Operator fix_residual_sugar Logs:\n",
      "stdout:\n",
      "            missing residual sugar values: 500\n",
      "\n",
      "            --------------------------\n",
      "            stderr:\n",
      "\n",
      "\n",
      "Operator predict_quality Logs:\n",
      "stdout:\n",
      "            Feature Columns Index(['alcohol', 'chlorides', 'citric_acid', 'density', 'fixed_acidity',\n",
      "       'free_sulfur_dioxide', 'is_red', 'ph', 'residual_sugar', 'sulphates',\n",
      "       'total_sulfur_dioxide', 'volatile_acidity'],\n",
      "      dtype='object')\n",
      "\n",
      "            --------------------------\n",
      "            stderr:\n",
      "\n",
      "\n"
     ]
    },
    {
     "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>id</th>\n",
       "      <th>color</th>\n",
       "      <th>fixed_acidity</th>\n",
       "      <th>volatile_acidity</th>\n",
       "      <th>citric_acid</th>\n",
       "      <th>residual_sugar</th>\n",
       "      <th>chlorides</th>\n",
       "      <th>free_sulfur_dioxide</th>\n",
       "      <th>total_sulfur_dioxide</th>\n",
       "      <th>density</th>\n",
       "      <th>ph</th>\n",
       "      <th>sulphates</th>\n",
       "      <th>alcohol</th>\n",
       "      <th>quality</th>\n",
       "      <th>is_red</th>\n",
       "      <th>pred_quality</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>white</td>\n",
       "      <td>7.0</td>\n",
       "      <td>0.17</td>\n",
       "      <td>0.74</td>\n",
       "      <td>10.027690</td>\n",
       "      <td>0.045</td>\n",
       "      <td>24.0</td>\n",
       "      <td>126.0</td>\n",
       "      <td>0.99420</td>\n",
       "      <td>3.26</td>\n",
       "      <td>0.38</td>\n",
       "      <td>12.2</td>\n",
       "      <td>8.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>6.653566</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>red</td>\n",
       "      <td>7.7</td>\n",
       "      <td>0.64</td>\n",
       "      <td>0.21</td>\n",
       "      <td>2.200000</td>\n",
       "      <td>0.077</td>\n",
       "      <td>32.0</td>\n",
       "      <td>133.0</td>\n",
       "      <td>0.99560</td>\n",
       "      <td>3.27</td>\n",
       "      <td>0.45</td>\n",
       "      <td>9.9</td>\n",
       "      <td>5.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>5.569721</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2</td>\n",
       "      <td>white</td>\n",
       "      <td>6.8</td>\n",
       "      <td>0.39</td>\n",
       "      <td>0.34</td>\n",
       "      <td>7.400000</td>\n",
       "      <td>0.020</td>\n",
       "      <td>38.0</td>\n",
       "      <td>133.0</td>\n",
       "      <td>0.99212</td>\n",
       "      <td>3.18</td>\n",
       "      <td>0.44</td>\n",
       "      <td>12.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>6.653566</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>3</td>\n",
       "      <td>white</td>\n",
       "      <td>6.3</td>\n",
       "      <td>0.28</td>\n",
       "      <td>0.47</td>\n",
       "      <td>10.697218</td>\n",
       "      <td>0.040</td>\n",
       "      <td>61.0</td>\n",
       "      <td>183.0</td>\n",
       "      <td>0.99592</td>\n",
       "      <td>3.12</td>\n",
       "      <td>0.51</td>\n",
       "      <td>9.5</td>\n",
       "      <td>6.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>5.255291</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>4</td>\n",
       "      <td>white</td>\n",
       "      <td>7.4</td>\n",
       "      <td>0.35</td>\n",
       "      <td>0.20</td>\n",
       "      <td>13.900000</td>\n",
       "      <td>0.054</td>\n",
       "      <td>63.0</td>\n",
       "      <td>229.0</td>\n",
       "      <td>0.99888</td>\n",
       "      <td>3.11</td>\n",
       "      <td>0.50</td>\n",
       "      <td>8.9</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>5.255291</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   id  color  fixed_acidity  volatile_acidity  citric_acid  residual_sugar  \\\n",
       "0   0  white            7.0              0.17         0.74       10.027690   \n",
       "1   1    red            7.7              0.64         0.21        2.200000   \n",
       "2   2  white            6.8              0.39         0.34        7.400000   \n",
       "3   3  white            6.3              0.28         0.47       10.697218   \n",
       "4   4  white            7.4              0.35         0.20       13.900000   \n",
       "\n",
       "   chlorides  free_sulfur_dioxide  total_sulfur_dioxide  density    ph  \\\n",
       "0      0.045                 24.0                 126.0  0.99420  3.26   \n",
       "1      0.077                 32.0                 133.0  0.99560  3.27   \n",
       "2      0.020                 38.0                 133.0  0.99212  3.18   \n",
       "3      0.040                 61.0                 183.0  0.99592  3.12   \n",
       "4      0.054                 63.0                 229.0  0.99888  3.11   \n",
       "\n",
       "   sulphates  alcohol  quality  is_red  pred_quality  \n",
       "0       0.38     12.2      8.0     0.0      6.653566  \n",
       "1       0.45      9.9      5.0     1.0      5.569721  \n",
       "2       0.44     12.0      NaN     0.0      6.653566  \n",
       "3       0.51      9.5      6.0     0.0      5.255291  \n",
       "4       0.50      8.9      NaN     0.0      5.255291  "
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "predicted_quality = predict_quality(featurized_wines)\n",
    "predicted_quality.get().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e686bcf6-d417-4455-bc56-17a98cebb29d",
   "metadata": {},
   "source": [
    "## Checking Our Predictions\n",
    "\n",
    "As a sanity check, we also verify that the wine quality predictions are reasonable. We'll do this by defining another `metric` on the `predicted_quality` table that calculates the RMSE of the predictions for the wines for which we have actual labels."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "5d8320b4-ff30-4e54-944c-bcfc2c7f3187",
   "metadata": {},
   "outputs": [],
   "source": [
    "@metric()\n",
    "def get_rmse(df):\n",
    "    \"\"\"\n",
    "    This metric function takes in a DataFrame and assumes it has two columns,\n",
    "    `quality` and `pred_quality`. It uses numpy to calculate the root mean squared\n",
    "    error of the predicted quality column. It ignores any rows for which the quality\n",
    "    column does not have a valid value.\n",
    "    \"\"\"\n",
    "    import numpy as np\n",
    "\n",
    "    residuals = (df[\"quality\"] - df[\"pred_quality\"]).dropna()\n",
    "    print(\"Computing error using:\", len(residuals), \"rows.\")\n",
    "    return np.sqrt((residuals * residuals).mean())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "a1ae3f51-49a9-4a49-809e-4274eb6848fe",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Operator fix_residual_sugar Logs:\n",
      "stdout:\n",
      "            missing residual sugar values: 500\n",
      "\n",
      "            --------------------------\n",
      "            stderr:\n",
      "\n",
      "\n",
      "Operator predict_quality Logs:\n",
      "stdout:\n",
      "            Feature Columns Index(['alcohol', 'chlorides', 'citric_acid', 'density', 'fixed_acidity',\n",
      "       'free_sulfur_dioxide', 'is_red', 'ph', 'residual_sugar', 'sulphates',\n",
      "       'total_sulfur_dioxide', 'volatile_acidity'],\n",
      "      dtype='object')\n",
      "\n",
      "            --------------------------\n",
      "            stderr:\n",
      "\n",
      "\n",
      "Operator get_rmse Logs:\n",
      "stdout:\n",
      "            Computing error using: 5997 rows.\n",
      "\n",
      "            --------------------------\n",
      "            stderr:\n",
      "\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "0.7440468072891235"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rmse = get_rmse(predicted_quality)\n",
    "rmse.get()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "aaf5059e-f8f6-4de8-becc-6499e99540f0",
   "metadata": {},
   "source": [
    "Here we set two bounds on the error.  The first is a warning and the second we set to an error to avoid rendering potentially egregiously bad predictions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "550cd7b8-26c5-4107-a87e-cafa5e04a493",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<aqueduct.check_artifact.CheckArtifact at 0x13451cd30>"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# We set two bounds on the RMSE. If the RMSE is above 1.0,\n",
    "# we will get an error, but we will not fail the workflow;\n",
    "# however, if the RMSE is above 3.0, we will fail the whole\n",
    "# workflow.\n",
    "rmse.bound(upper=1.0)\n",
    "rmse.bound(upper=3.0, severity=\"error\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "21c5c7f7-f67d-46e4-b0e6-00647b271ea4",
   "metadata": {},
   "source": [
    "## Saving the Predicted Wine Quality\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "f2803925-aeb7-43b6-adae-efb798c36f1e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# This tells Aqueduct to save the results in predicted_quality\n",
    "# back to the demo DB we configured earlier.\n",
    "# NOTE: At this point, no data is actually saved! This is just\n",
    "# part of a workflow spec that will be executed once the workflow\n",
    "# is published in the next cell.\n",
    "demodb.save(predicted_quality, table_name=\"pred_wine_quality\", update_mode=\"replace\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "970d79e3-0640-4ea6-a5db-58bec79e4373",
   "metadata": {},
   "source": [
    "## Publishing the Workflow\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "24d520c2-34a4-4521-886f-278b051cff66",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Url:  http://localhost:8080/workflow/0361f939-ad4e-4b3c-8f7d-f87d4a15f50f\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<aqueduct.flow.Flow at 0x134430fa0>"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# This publishes all of the logic needed to create predicted_quality\n",
    "# rmse, and num_labeled to Aqueduct. The URL below will take you to the\n",
    "# Aqueduct UI, which will show you the status of your workflow\n",
    "# runs and allow you to inspect them.\n",
    "from textwrap import dedent\n",
    "\n",
    "client.publish_flow(\n",
    "    \"Wine Ratings\",\n",
    "    dedent(\n",
    "        \"\"\"\n",
    "        This workflow builds a model to predict missing ratings for wines \n",
    "        and then uses that model to fill in missing ratings.\n",
    "        \"\"\"\n",
    "    ),\n",
    "    # Uncomment the following line to schedule on a daily basis.\n",
    "    # schedule=aqueduct.daily(),\n",
    "    artifacts=[predicted_quality, rmse, num_labeled],\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b60862cf",
   "metadata": {},
   "source": [
    "Clicking on the URL above will take you to the Aqueudct UI where you can see the workflow that we just created! On the Aqueduct UI, you'll be able to see the DAG of operators we just created, click into any of those operators, and see the data and metadata associated with each stage of the pipeline."
   ]
  }
 ],
 "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.8.2"
  },
  "vscode": {
   "interpreter": {
    "hash": "31f2aee4e71d21fbe5cf8b01ff0e069b9275f58929596ceb00d14d90e3e16cd6"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}