05_Data_Science.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import snowflake.snowpark as snp\n",
    "from snowflake.snowpark.functions import sproc\n",
    "import joblib\n",
    "from sklearn.preprocessing import OneHotEncoder\n",
    "from sklearn.compose import ColumnTransformer\n",
    "from sklearn.pipeline import Pipeline\n",
    "from kmodes.kmodes import KModes\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import io"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Science\n",
    "\n",
    "Now that the data engineers have loaded and transformed the data to the `squirrels_engineered_features` table, we can begin our model development. For this, we will leverage Snowpark to do **clustering analysis**.  \n",
    "<br> \n",
    "The Snowpark Python client-side Dataframe API allows us to push-down most of the computation for preparation and feature engineering to Snowpark. For security and governance reasons we can read data into memory for model training and inference but no intermediate data products will be stored outside of Snowflake. \n",
    "Also, since training a model usually requires more intensive computation we are going to use a larger warehouse for this task. Snowflake supports resizing a warehouse at any time, you can scale up a warehouse in seconds or less. For this exercise, we are going to use the data science warehouse, so that every team can work completely independent of the others."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1. Load the Credentials"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from steps.snowpark_connection import snowpark_connect\n",
    "session, state_dict = snowpark_connect('./include/state.json')\n",
    "session.use_warehouse(state_dict['compute_parameters']['ds_warehouse'])\n",
    "print(session.sql('select current_warehouse(), current_database(), current_schema()').collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. Create stages to save the ML model/pipeline and permanent UDFs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create stage to store models\n",
    "query = \"create or replace stage models\" +\\\n",
    "        \" directory = (enable = true)\" +\\\n",
    "        \" copy_options = (on_error='skip_file')\"\n",
    "        \n",
    "session.sql(query).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create stage to store udfs\n",
    "query = \"create or replace stage udf\" +\\\n",
    "        \" copy_options = (on_error='skip_file')\"\n",
    "        \n",
    "session.sql(query).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# clear out existing packages and imports\n",
    "session.clear_packages()\n",
    "session.clear_imports()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# let the snowflake server know what packages to manage on your behalf\n",
    "session.add_packages('snowflake-snowpark-python', 'scikit-learn', 'pandas', 'numpy', 'joblib', 'cachetools')\n",
    "session.add_import('./include/kmodes.zip')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# save artifacts to stages\n",
    "def save_file(session, model, path):\n",
    "  input_stream = io.BytesIO()\n",
    "  joblib.dump(model, input_stream)\n",
    "  session._conn._cursor.upload_stream(input_stream, path)\n",
    "  return \"successfully created file: \" + path"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3. Train model using Kmodes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Snowpark server-side Anaconda runtime has a large [list of Python modules included](https://docs.snowflake.com/en/developer-guide/udf/python/udf-python-packages.html) for our UDF.  However, the data scientist built this code based on kmodes which is not currently in the Snowpark distribution.\n",
    "  \n",
    "  We can simply add [kmodes](https://github.com/nicodv/kmodes), as well as our own team's python code, as import dependencies."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# train model using engineered features from a snowflake table\n",
    "def train_model(session: snp.Session) -> float:\n",
    "    snowdf = session.table(\"squirrels_engineered_features\")\n",
    "    train = snowdf.select(\"HAS_ACTIVITY\", \"IS_STRESSED\", \"COMFY_WITH_HUMANS\").to_pandas()\n",
    "    preprocessor = ColumnTransformer([\n",
    "        (\"cat\", OneHotEncoder(), train.columns)])\n",
    "    full_pipeline = Pipeline([\n",
    "            ('preprocessor', preprocessor),\n",
    "            ('model', KModes(n_clusters=3, init = \"Huang\", random_state=0)),\n",
    "        ])\n",
    "\n",
    "    full_pipeline.fit(train)\n",
    "    # save the full pipeline including the model\n",
    "    save_file(session, full_pipeline, \"@MODELS/kmodes.joblib\")\n",
    "    fitClusters = full_pipeline.predict(train)\n",
    "\n",
    "    return len(np.unique(fitClusters))\n",
    "\n",
    "# Create an instance of StoredProcedure using the sproc() function\n",
    "train_model_sp = sproc(train_model, replace=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "¯\\\\\\_(ツ)_/¯ ```Clustering``` is an unsupervised learning method whose task is to divide the population or data points into a number of groups, such that data points in a group are more similar to other data points in the same group and dissimilar to the data points in other groups. It is basically a collection of objects based on similarity and dissimilarity between them.\n",
    "\n",
    "```KModes``` clustering is one of the unsupervised Machine Learning algorithms that is used to cluster categorical variables (no intrinsic order to the categories).\n",
    "\n",
    "You might be wondering, why ```KModes``` when we already have ```KMeans```.\n",
    "\n",
    "KMeans uses mathematical measures (distance) to cluster continuous data. The lesser the distance, the more similar our data points are. Centroids are updated by Means.\n",
    "But for categorical data points, we cannot calculate the distance. So we go for KModes algorithm. It uses the dissimilarities (total mismatches) between the data points. The lesser the dissimilarities the more similar our data points are. It uses Modes instead of means. The mode is the number that occurs most often in a data set."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4. Run the training within the SPROC"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# train model direcly on the snowflake server: the code is moved to data!\n",
    "train_model_sp()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5. Model/Pipeline Deployment using UDF\n",
    "Here we can use Snowpark User Defined Functions for inference without having to pull data out of Snowflake."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "import cachetools\n",
    "import os\n",
    "from snowflake.snowpark.functions import udf\n",
    "session.add_import(\"@MODELS/kmodes.joblib\")  \n",
    "\n",
    "# get model from stage\n",
    "@cachetools.cached(cache={})\n",
    "def read_file(filename):\n",
    "       import_dir = sys._xoptions.get(\"snowflake_import_directory\")\n",
    "       if import_dir:\n",
    "              with open(os.path.join(import_dir, filename), 'rb') as file:\n",
    "                     m = joblib.load(file)\n",
    "                     return m\n",
    "\n",
    "features = [\"HAS_ACTIVITY\", \"IS_STRESSED\", \"COMFY_WITH_HUMANS\"]\n",
    "\n",
    "# create udf to apply predictions on new data\n",
    "@udf(name=\"predict\", is_permanent=True, stage_location=\"@udf\", replace=True)\n",
    "def predict(HAS_ACTIVITY: int, IS_STRESSED: int, COMFY_WITH_HUMANS: int) -> int:\n",
    "       m = read_file('kmodes.joblib')       \n",
    "       row = pd.DataFrame([locals()], columns=features)\n",
    "       return m.predict(row)[0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets check how the model performs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# apply udf to make predictions on new data\n",
    "snowdf_test = session.table(\"squirrels_engineered_features\")\n",
    "inputs = snowdf_test.select(\"HAS_ACTIVITY\", \"IS_STRESSED\", \"COMFY_WITH_HUMANS\")\n",
    "snowdf_results = snowdf_test.select(*snowdf_test.columns,\n",
    "                    predict(*inputs).alias('CLUSTER'), \n",
    "                    )\n",
    "                    \n",
    "snowdf_results.to_pandas().head(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# save clusters back to snowflake\n",
    "snowdf_results.write.mode(\"overwrite\").save_as_table(\"squirrels_engineered_clusters\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "session.close()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.8.13 ('squirrelly_env')",
   "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.13"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "2ba27bfd1749f7c3a7bc0f0e0b81a34114b55e33b7aa6e17a375be311f1d8668"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}