training.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Nuclio - Training function"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Environment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuclio: ignore\n",
    "import nuclio"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Function"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Utils\n",
    "import os\n",
    "import time\n",
    "import yaml\n",
    "import pandas as pd\n",
    "import datetime\n",
    "import itertools\n",
    "import pickle\n",
    "\n",
    "# DB Connection\n",
    "import v3io_frames as v3f\n",
    "\n",
    "# Parallelization\n",
    "import dask.dataframe as dd\n",
    "from dask.distributed import Client\n",
    "\n",
    "# Function\n",
    "import dask_ml.model_selection as dcv\n",
    "import xgboost as xgb\n",
    "from mlrun import get_or_create_ctx"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[mlrun] 2020-03-10 18:07:07,814 logging run results to: http://10.233.60.111:8080\n"
     ]
    }
   ],
   "source": [
    "mlruncontext = get_or_create_ctx('training')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# Netops features table\n",
    "features_table=os.path.join(str(mlruncontext.get_param('NETAPP_MOUNT_PATH')),mlruncontext.get_param('FEATURES_TABLE', 'netops_features'))\n",
    "mlruncontext.logger.info(\"FEATURES %s\"%features_table)\n",
    "# Set time to train on\n",
    "train_on_last = mlruncontext.get_param('TRAIN_ON_LAST', '7d')\n",
    "\n",
    "# Get saving configuration\n",
    "is_from_tsdb = (int(mlruncontext.get_param('FROM_TSDB', 0)) == 1)\n",
    "\n",
    " # Create saving directory if needed\n",
    "filepath = os.path.join(features_table)\n",
    "if not os.path.exists(filepath):\n",
    "    os.makedirs(filepath)\n",
    "\n",
    "# Set training set size\n",
    "train_set_size = float(mlruncontext.get_param('TRAIN_SIZE', 0.7))\n",
    "train_size = float(mlruncontext.get_param('TRAIN_SIZE', 0.7))\n",
    "\n",
    "# Dask shards / CV\n",
    "shards = int(mlruncontext.get_param('NUMBER_OF_SHARDS', 4))\n",
    "\n",
    "# Create save-to folder if needed\n",
    "model_filepath = os.path.join(mlruncontext.get_param('APP_DIR'),mlruncontext.get_param('SAVE_TO', '/v3io/bigdata/netops/models'))\n",
    "if not os.path.exists(model_filepath):\n",
    "    os.makedirs(model_filepath)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Helper functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_data_parquet():\n",
    "    # Get parquet files\n",
    "    mpath = [os.path.join(features_table, file) for file in os.listdir(features_table)]\n",
    "    \n",
    "    # Get latest filename\n",
    "    latest = max(mpath, key=os.path.getmtime)\n",
    "    print(latest)\n",
    "    context.logger.debug('Reading data from: %s'%latest)\n",
    "    \n",
    "    # Load parquet to dask\n",
    "    df = dd.read_parquet(latest)\n",
    "    \n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_train_test_sets_from_data(df: pd.DataFrame):\n",
    "    drop_columns = [col for col in df.columns if 'is_error' in col]\n",
    "    X = df.drop(drop_columns, axis=1)\n",
    "    y = df.loc[:, 'is_error']\n",
    "    X_train, X_test, y_train, y_test = dcv.train_test_split(X, y, train_size=train_size, test_size=1-train_size)\n",
    "    return X_train, X_test, y_train, y_test"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Handler"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def handler(context):\n",
    "    # Get data\n",
    "    df = get_data_parquet() \n",
    "\n",
    "    # Split to Train / Test datasets\n",
    "    X_train, X_test, y_train, y_test = get_train_test_sets_from_data(df)\n",
    "    \n",
    "    # Train\n",
    "    model = xgb.XGBClassifier()\n",
    "    model.fit(X_train, y_train)\n",
    "    \n",
    "    # Score\n",
    "    score = model.score(X_test, y_test)\n",
    "    context.log_result('accuracy',score)\n",
    "    # Save model\n",
    "    model_file = os.path.join(context.get_param('APP_DIR'),context.get_param('SAVE_TO'),context.get_param('MODEL_FILENAME'))\n",
    "    context.logger.info(\"Save model to %s\"%model_file)\n",
    "    context.log_artifact('trained_model.pickle',model_file)\n",
    "    pickle.dump(model, open(model_file,'wb'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#output = handler(context)\n",
    "#output"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Deployment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}