notebooks/preprocessor.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Pre-Processor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import nuclio"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "%nuclio: setting kind to 'nuclio'\n"
     ]
    }
   ],
   "source": [
    "%nuclio config kind = \"nuclio\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "%nuclio: setting 'aggregate_fn_url' environment variable\n",
      "%nuclio: setting 'METRICS_TABLE' environment variable\n",
      "%nuclio: setting 'FEATURES_TABLE' environment variable\n",
      "%nuclio: setting 'base_dataset' environment variable\n",
      "%nuclio: setting 'keys' environment variable\n",
      "%nuclio: setting 'metrics' environment variable\n",
      "%nuclio: setting 'metric_aggs' environment variable\n",
      "%nuclio: setting 'suffix' environment variable\n",
      "%nuclio: setting 'window' environment variable\n",
      "%nuclio: setting 'center' environment variable\n",
      "%nuclio: setting 'inplace' environment variable\n",
      "%nuclio: setting 'drop_na' environment variable\n",
      "%nuclio: setting 'files_to_select' environment variable\n",
      "%nuclio: setting 'label_col' environment variable\n",
      "%nuclio: setting 'is_save_to_tsdb' environment variable\n"
     ]
    }
   ],
   "source": [
    "%%nuclio env\n",
    "\n",
    "aggregate_fn_url = /User/functions/aggregate/function.yaml\n",
    "METRICS_TABLE = /User/demo-network-operations/data\n",
    "FEATURES_TABLE = /User/demo-network-operations/features\n",
    "\n",
    "base_dataset = /User/demo-network-operations/artifacts/selected_features.parquet\n",
    "\n",
    "keys = timestamp,company,data_center,device\n",
    "metrics = [\"cpu_utilization\", \"throughput\", \"packet_loss\", \"latency\"]\n",
    "metric_aggs = [\"mean\", \"sum\", \"std\", \"var\", \"min\", \"max\", \"median\"]\n",
    "suffix = daily\n",
    "window = 3\n",
    "center = 0\n",
    "inplace = 0\n",
    "drop_na = 1\n",
    "files_to_select = 1\n",
    "label_col = is_error\n",
    "\n",
    "is_save_to_tsdb = 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# %%nuclio env -c\n",
    "\n",
    "# aggregate_fn_url = hub://aggregate\n",
    "# METRICS_TABLE = network-operations/metrics\n",
    "# FEATURES_TABLE = network-operations/features\n",
    "\n",
    "# keys = company,data_center,device\n",
    "# metrics = timestamp, cpu_utilization,latency,packet_loss,throughput\n",
    "# metric_aggs = mean,max\n",
    "# suffix = daily\n",
    "# window = 3\n",
    "# center = 0\n",
    "# inplace = 1\n",
    "# drop_na = 1\n",
    "\n",
    "# is_save_to_tsdb = 0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuclio: start-code"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import pandas as pd\n",
    "from mlrun.datastore import DataItem\n",
    "import ast\n",
    "\n",
    "from typing import Union\n",
    "from mlrun import mlconf, import_function, mount_v3io, NewTask, function_to_module, get_or_create_ctx\n",
    "from mlrun.run import get_dataitem"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_data_tsdb(context):\n",
    "    df = context.v3f.read(backend='tsdb', query=f'select cpu_utilization, latency, packet_loss, throughput, is_error from {context.metrics_table}',\n",
    "                          start=f'now-2h', end='now', multi_index=True)\n",
    "    df = format_df_from_tsdb(context, df)\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_data_parquet(context):\n",
    "    mpath = [os.path.join(context.metrics_table, file) for file in os.listdir(context.metrics_table) if file.endswith(('parquet', 'pq'))]\n",
    "    files_by_updated = sorted(mpath, key=os.path.getmtime, reverse=True)\n",
    "    context.logger.info(files_by_updated)\n",
    "    latest = files_by_updated[:context.files_to_select]\n",
    "    context.logger.info(f'Aggregating {latest}')\n",
    "    input_df = pd.concat([pd.read_parquet(df) for df in latest])\n",
    "    return input_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def save_to_tsdb(context, features: pd.DataFrame):   \n",
    "    context.v3f.write('tsdb', context.features_table, features)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def save_to_parquet(context, df: pd.DataFrame):\n",
    "    print('Saving features to Parquet')\n",
    "    \n",
    "    # Need to fix timestamps from ns to ms if we write to parquet\n",
    "    df = df.reset_index()\n",
    "    df['timestamp'] = df.loc[:, 'timestamp'].astype('datetime64[ms]')\n",
    "    \n",
    "    # Fix indexes\n",
    "    df = df.set_index(context.keys)\n",
    "    \n",
    "    # Save parquet\n",
    "    first_timestamp = df.index[0][0].strftime('%Y%m%dT%H%M%S')\n",
    "    last_timestamp = df.index[-1][0].strftime('%Y%m%dT%H%M%S')\n",
    "    filename = first_timestamp + '-' + last_timestamp + '.parquet'\n",
    "    filepath = os.path.join(context.features_table, filename)\n",
    "    with open(filepath, 'wb+') as f:\n",
    "        df.to_parquet(f)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "def init_context(context):\n",
    "    \n",
    "    mlconf.dbpath = 'http://mlrun-api:8080'\n",
    "    \n",
    "    # Setup aggregate function\n",
    "    aggregate_fn = import_function(os.getenv('aggregate_fn_url', 'hub://aggregate'))\n",
    "    mod = function_to_module(aggregate_fn)\n",
    "    setattr(context, 'aggregate', mod.aggregate)\n",
    "    \n",
    "    ag_context = get_or_create_ctx('aggregate')\n",
    "    setattr(context, 'mlrun_ctx', ag_context)\n",
    "    \n",
    "    # Set vars from env\n",
    "    setattr(context, 'metrics_table', os.getenv('METRICS_TABLE', 'netops_metrics'))\n",
    "    setattr(context, 'features_table', os.getenv('FEATURES_TABLE', 'netops_features'))\n",
    "    setattr(context, 'keys', os.getenv('keys', '').split(','))\n",
    "    setattr(context, 'metrics', ast.literal_eval(os.getenv('metrics', '')))\n",
    "    setattr(context, 'metric_aggs', ast.literal_eval(os.getenv('metric_aggs', '')))\n",
    "    setattr(context, 'suffix', os.getenv('suffix', '_agg'))\n",
    "    setattr(context, 'window', int(os.getenv('window', '3')))\n",
    "    setattr(context, 'center', bool(int(os.getenv('center', '0'))))\n",
    "    setattr(context, 'inplace', bool(int(os.getenv('inplace', '0'))))\n",
    "    setattr(context, 'drop_na', bool(int(os.getenv('drop_na', '1'))))\n",
    "    setattr(context, 'files_to_select', int(os.getenv('files_to_select', 1)))\n",
    "    \n",
    "    sample_dataset = get_dataitem(os.environ['base_dataset']).as_df()\n",
    "    selected_features = [col for col in list(sample_dataset.columns) if col != os.getenv('label_col', '')]\n",
    "    aggregated_features = [feature.split('_')[:-1] for feature in selected_features if feature.endswith(context.suffix)]\n",
    "    base_features = set([f[0] for f in aggregated_features])\n",
    "    aggregations = set([f[1] for f in aggregated_features])\n",
    "    setattr(context, 'features', selected_features)\n",
    "    setattr(context, 'base_features', base_features)\n",
    "    setattr(context, 'aggregations', aggregations)\n",
    "    \n",
    "    \n",
    "    \n",
    "    # Save to TSDB\n",
    "    is_save_to_tsdb = bool(int(os.getenv('save_to_tsdb', '0')))\n",
    "    if is_save_to_tsdb:\n",
    "        # Create our DB client\n",
    "        v3io_client = v3f.Client(address='framesd:8081', container='bigdata')\n",
    "        setattr(context, 'v3f', v3io_client)\n",
    "        \n",
    "        # Create features table if neede\n",
    "        context.v3f.create('tsdb', context.features_table, attrs={'rate': '1/s'}, if_exists=1)\n",
    "        \n",
    "        # Set TSDB reading function\n",
    "        setattr(context, 'read', get_data_tsdb)\n",
    "        \n",
    "        # Set TSDB saving function\n",
    "        setattr(context, 'write', save_to_tsdb)\n",
    "        \n",
    "    # Save to Parquet\n",
    "    else:\n",
    "         # Create saving directory if needed\n",
    "        filepath = os.path.join(context.features_table)\n",
    "        if not os.path.exists(filepath):\n",
    "            os.makedirs(filepath)\n",
    "            \n",
    "        # Set Parquet reading function\n",
    "        setattr(context, 'read', get_data_parquet)\n",
    "        \n",
    "        # Set Parquet saving function\n",
    "        setattr(context, 'write', save_to_parquet)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "def handler(context, event):\n",
    "    \n",
    "    # Get latest parquets\n",
    "    df = context.read(context)\n",
    "    \n",
    "    # Call aggregate\n",
    "    res = context.aggregate(context=context.mlrun_ctx,\n",
    "              df_artifact=df,\n",
    "              save_to=context.features_table, \n",
    "              keys=context.keys, \n",
    "              metrics=context.metrics, \n",
    "              metric_aggs=context.metric_aggs, \n",
    "              suffix=context.suffix, \n",
    "              window=context.window, \n",
    "              center=context.center, \n",
    "              inplace=context.inplace,\n",
    "              drop_na=context.drop_na)\n",
    "    \n",
    "    context.logger.info(f'res.columns: {res.columns}')\n",
    "    context.logger.info(f'context.columns: {context.features}')\n",
    "    res = res[context.features]\n",
    "    \n",
    "    # Save\n",
    "    context.write(context, res)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuclio: end-code"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Local test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 223,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[mlrun] 2020-07-08 12:47:04,161 logging run results to: http://mlrun-api:8080\n"
     ]
    }
   ],
   "source": [
    "init_context(context)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "event = nuclio.Event(body='')\n",
    "out = handler(context, event)\n",
    "out"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "from mlrun import code_to_function, mount_v3io"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<mlrun.runtimes.function.RemoteRuntime at 0x7fb7401972b0>"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fn = code_to_function('nuclio-preprocessor',\n",
    "                      kind='nuclio',\n",
    "                      project='network-operations')\n",
    "fn.spec.base_spec['spec']['build']['baseImage'] = 'mlrun/ml-models'\n",
    "fn.apply(mount_v3io())\n",
    "fn.add_trigger('cron', nuclio.triggers.CronTrigger(interval='1m'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "> 2020-07-27 06:47:07,600 [debug] saving function: nuclio-preprocessor, tag: \n",
      "> 2020-07-27 06:47:07,659 [info] function spec saved to path: ../src/preprocessor.yaml\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<mlrun.runtimes.function.RemoteRuntime at 0x7fb7401972b0>"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fn.save()\n",
    "fn.export('../src/preprocessor.yaml')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 202,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[mlrun] 2020-07-08 12:37:26,608 deploy started\n",
      "[nuclio] 2020-07-08 12:37:28,726 (info) Build complete\n",
      "[nuclio] 2020-07-08 12:37:40,864 (info) Function deploy complete\n",
      "[nuclio] 2020-07-08 12:37:40,870 done updating network-operations-nuclio-preprocessor, function address: 192.168.224.209:31857\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'http://192.168.224.209:31857'"
      ]
     },
     "execution_count": 202,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fn.deploy(project='network-operations')"
   ]
  },
  {
   "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
}