concept_drift_streaming/concept_drift_streaming.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Concept Drift Streaming"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import nuclio"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pprint import pprint"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%nuclio cmd -c\n",
    "python -m pip install scikit-multiflow==0.4.1\n",
    "python -m pip install v3io_frames"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "%nuclio: setting kind to 'nuclio'\n",
      "%nuclio: setting spec.build.baseImage to 'mlrun/ml-models'\n"
     ]
    }
   ],
   "source": [
    "# Define function spec\n",
    "%nuclio config kind = \"nuclio\"\n",
    "%nuclio config spec.build.baseImage = \"mlrun/ml-models\"\n",
    "\n",
    "# Add V3IO Mount\n",
    "# %nuclio env %v3io"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuclio: ignore\n",
    "env = {'label_col': 'resp',\n",
    "       'prediction_col': 'prediction',\n",
    "       'drift_stream': '/bigdata/network-operations/drift_stream',\n",
    "       'tsdb_table': 'network-operations/drift_tsdb',\n",
    "       'pagehinkley_threshold': 10,\n",
    "       'models': ['pagehinkley', 'ddm', 'eddm'],\n",
    "       'window_size': 10}\n",
    "config = {'kind': 'nuclio',\n",
    "          'spec.build.baseImage': 'mlrun/ml-models'}\n",
    "cmd = ['python -m pip install scikit-multiflow',\n",
    "       'python -m pip install v3io_frames']\n",
    "v3io = True\n",
    "config = nuclio.ConfigSpec(env=env,\n",
    "                           config=config,\n",
    "                           cmd=cmd,\n",
    "                           v3io=v3io)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuclio: start-code"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "import skmultiflow.drift_detection\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import os\n",
    "import json\n",
    "import v3io.dataplane\n",
    "import v3io_frames as v3f\n",
    "import requests\n",
    "from cloudpickle import load\n",
    "\n",
    "# For testing\n",
    "import random"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def split_path(mntpath=''):\n",
    "    if mntpath[0] == '/':\n",
    "        mntpath = mntpath[1:]\n",
    "    paths = mntpath.split('/')\n",
    "    container = paths[0]\n",
    "    subpath = ''\n",
    "    if len(paths) > 1:\n",
    "        subpath = mntpath[len(container):]\n",
    "    return container, subpath\n",
    "\n",
    "\n",
    "def create_stream(context, path, shards=1):\n",
    "    # create a stream w/8 shards\n",
    "    container, stream_path = split_path(path)\n",
    "    context.logger.info(f'Creating stream in Container: {container} & Path {stream_path}')\n",
    "    response = context.v3io_client.create_stream(container=container,\n",
    "                                        path=stream_path, \n",
    "                                        shard_count=shards,\n",
    "                                        raise_for_status=v3io.dataplane.RaiseForStatus.never)\n",
    "    response.raise_for_status([409, 204])\n",
    "    \n",
    "    \n",
    "def push_to_stream(context, stream_path, data):\n",
    "    records = [{'data': json.dumps(rec)} for rec in data]\n",
    "    container, stream_path = split_path(stream_path)\n",
    "    response = context.v3io_client.put_records(container=container,\n",
    "                                               path=stream_path, \n",
    "                                               records=records)\n",
    "\n",
    "\n",
    "def construct_record(record):\n",
    "    label_col = os.getenv('label_col', 'label')\n",
    "    prediction_col = os.getenv('prediction_col', 'prediction')\n",
    "    res = dict([(k, record[k]) for k in ['when', 'class', 'model', 'resp', 'request']])\n",
    "    res['feature_vector'] = res.pop('request')['instances'][0]\n",
    "    res['timestamp'] = res.pop('when')\n",
    "    res['prediction'] = res['resp'][0]\n",
    "    return res"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "def init_context(context):\n",
    "    # create a v3io context object\n",
    "    v3io_client = v3io.dataplane.Client()\n",
    "    setattr(context, \"v3io_client\", v3io_client)\n",
    "    \n",
    "    # Setup windowing for TSDB writer\n",
    "    v3f_client = v3f.Client('framesd:8081', container='bigdata')\n",
    "    setattr(context, \"v3f\", v3f_client)\n",
    "    window = []\n",
    "    setattr(context, 'window', window)\n",
    "    setattr(context, 'window_size', int(os.getenv('window_size', 10)))\n",
    "    setattr(context, 'tsdb_table', os.getenv('tsdb_table', 'concept_drift_tsdb_1'))\n",
    "    try:\n",
    "        context.v3f.create('tsdb', context.tsdb_table, rate='1/s', if_exists=1)\n",
    "    except Exception as e:\n",
    "        context.logger.info(f'Creating context with rate= faile for {e}')\n",
    "        context.v3f.create('tsdb', context.tsdb_table, attrs={'rate': '1/s'}, if_exists=1)\n",
    "    \n",
    "    # Setup callbacks\n",
    "    callbacks = [callback.strip() for callback in os.getenv('callbacks', '').split(',')]\n",
    "    setattr(context, 'callbacks', callbacks)\n",
    "    \n",
    "    # Setup drift stream\n",
    "    setattr(context, 'drift_stream', os.getenv('drift_stream', '/bigdata/drift_stream'))\n",
    "    try:\n",
    "        create_stream(context, context.drift_stream, int(os.getenv('drift_stream_shards', 1)))\n",
    "    except:\n",
    "        context.logger.info(f'{context.drift_stream} already exists')\n",
    "    \n",
    "    # Load models\n",
    "    models = {}\n",
    "    model_types = ['pagehinkely', 'ddm', 'eddm']\n",
    "    path_suffix = '_model_path'\n",
    "    for model in model_types:\n",
    "        model_env = f'{model}{path_suffix}'\n",
    "        if model_env in os.environ:\n",
    "            with open(os.environ[model_env], 'rb') as f:\n",
    "                models[model] = load(f)\n",
    "    setattr(context, 'models', models)\n",
    "    \n",
    "    # Columns to check\n",
    "    setattr(context, 'label_col', os.getenv('label_col', 'label'))\n",
    "    setattr(context, 'prediction_col', os.getenv('prediction_col', 'prediction'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "def handler(context, event):\n",
    "    # Construct event\n",
    "    context.logger.info(f'event: {event.body}')\n",
    "    full_event = json.loads(event.body)\n",
    "    record = construct_record(full_event)\n",
    "    \n",
    "    # Is our prediction wrong?\n",
    "    is_error = record[context.label_col] != record[context.prediction_col]\n",
    "    context.logger.info(f'Adding {is_error}')\n",
    "    \n",
    "    # Process the {is_error} element with our algorithms\n",
    "    for name, model in context.models.items():\n",
    "        # Add element\n",
    "        results = {'timestamp': record['timestamp']}\n",
    "        results['algorithm'] = name\n",
    "        model.add_element(is_error)\n",
    "        \n",
    "        # Detect warning zone (if applicable to the algorithm)\n",
    "        if hasattr(model, 'detected_warning_zone') and model.detected_warning_zone():\n",
    "            context.logger.info(f'{name}\\tWarning zone detected')\n",
    "            results['warning_zone'] = 1\n",
    "            full_event[f'{name}_warning_zone'] = 1\n",
    "        else:\n",
    "            results['warning_zone'] = 0\n",
    "            full_event[f'{name}_warning_zone'] = 0\n",
    "        \n",
    "        # Detect drift\n",
    "        if model.detected_change():\n",
    "            context.logger.info('Change Detected')\n",
    "            results['change_detected'] = 1\n",
    "            full_event[f'{name}_drift'] = 1\n",
    "        else:\n",
    "            results['change_detected'] = 0\n",
    "            full_event[f'{name}_drift'] = 0\n",
    "        context.window.append(results)\n",
    "    \n",
    "    # Return results\n",
    "    # Write to stream\n",
    "    push_to_stream(context, context.drift_stream, [full_event])\n",
    "    \n",
    "    # Add to callbacks\n",
    "    if context.callbacks != ['']:\n",
    "        for callback in context.callbacks:\n",
    "            requests.post(url=callback,\n",
    "                          json=full_event)\n",
    "    \n",
    "    if (len(context.window) / len(context.models)) >= context.window_size:\n",
    "        df = pd.DataFrame(context.window)\n",
    "        df['timestamp'] = pd.to_datetime(df['timestamp'])\n",
    "        df = df.set_index(['timestamp', 'algorithm'])\n",
    "        context.v3f.write('tsdb', context.tsdb_table, df)\n",
    "        context.window = []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuclio: end-code"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Test "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "init_context(context)\n",
    "event = nuclio.Event(body=json.dumps({'prediction': 0,\n",
    "                                      'when': 'now',\n",
    "                                      'class': 'ClassModel', \n",
    "                                      'model': 'tester_v1', \n",
    "                                      'resp': [0], \n",
    "                                      'request': {'instances': [[1, 1.2, 3]]}}))\n",
    "out = handler(context, event)\n",
    "out"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%nuclio deploy -n network-operations-concept-drift -p network-operations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Save function yaml"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "from os import path\n",
    "from mlrun import run_local, NewTask, mlconf, import_function, mount_v3io, code_to_function, get_run_db\n",
    "mlconf.dbpath = mlconf.dbpath or 'http://mlrun-api:8080'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[mlrun] 2020-07-14 13:49:22,720 function spec saved to path: /User/functions/concept_drift_streaming/function.yaml\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<mlrun.runtimes.function.RemoteRuntime at 0x7fb5a7de3e80>"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# create job function object from notebook code\n",
    "fn = code_to_function(\"concept_drift_streaming\", kind='nuclio')\n",
    "\n",
    "# add metadata (for templates and reuse)\n",
    "fn.spec.default_handler = \"handler\"\n",
    "fn.spec.description = \"Deploy a streaming Concept Drift detector on a labeled stream. the nuclio part of the concept_drift function\"\n",
    "fn.metadata.categories = [\"ml\", \"serve\"]\n",
    "fn.metadata.labels = {\"author\": \"orz\", \"framework\": \"sklearn\"}\n",
    "fn.export(\"/User/functions/concept_drift_streaming/function.yaml\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 120,
   "metadata": {},
   "outputs": [],
   "source": [
    "stream_trigger = nuclio.triggers.V3IOStreamTrigger(url='/bigdata/network-operations/inference_stream@cd2')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<mlrun.runtimes.function.RemoteRuntime at 0x7fa1dc063780>"
      ]
     },
     "execution_count": 121,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fn.add_trigger('labeled_stream', stream_trigger)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 122,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<mlrun.runtimes.function.RemoteRuntime at 0x7fa1dc063780>"
      ]
     },
     "execution_count": 122,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fn.apply(mount_v3io()).with_v3io()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fn.export(\"function.yaml\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Stream testing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "fn = import_function('./function.yaml')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fn.deploy(project='network-operations')"
   ]
  }
 ],
 "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
}