concept_drift/concept_drift.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Concept Drift - Deployer\n",
    "Deploy a streaming Concept Drift detector on a labeled stream.  \n",
    "It will initialize the selected drift detectors with the base_dataset's statistics and deploy the [concept_drift_streaming](https://github.com/mlrun/functions/blob/master/concept_drift_streaming/concept_drift_streaming.ipynb) function from the hub. <br>\n",
    "adding [V3IOStreamTrigger](https://nuclio.io/docs/latest/reference/triggers/v3iostream/) in order to listen to the input_stream."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **Steps**\n",
    "\n",
    "1. [Data exploration](#Data-exploration)\n",
    "2. [Creating the input stream](#Creating-the-input-stream)\n",
    "3. [Importing the function](#Importing-the-function)\n",
    "4. [Running the function remotely](#Running-the-function-remotely)\n",
    "5. [Testing the function](#Testing-the-function)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **Data exploration**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In order to know about the performance of a drift detector by measuring the different detection metrics, we need to know beforehand where a real drift occurs.<br>\n",
    "This is only possible with synthetic datasets.<br> The scikit-multiflow framework allows generating several kinds of synthetic data to simulate the occurrence of drifts.<br>\n",
    "[Harvard dataverse](https://dataverse.harvard.edu) provides futher explanations on the [used dataset](https://dataverse.harvard.edu/dataset.xhtml?persistentId=doi:10.7910/DVN/5OWRGB) along with different kinds of drifted datasets.<br>\n",
    "mixed_0101_abrupto has 4 concepts and 3 drifts at time steps 10000, 20000, and 30000.<br>\n",
    "Our dataset will be train-test-splitted, the train part (first 5000 examples) is used to train the model (that is generated easly using [sklearn_classifer](https://github.com/mlrun/functions/blob/master/sklearn_classifier/sklearn_classifier.ipynb)). <br>\n",
    "The test part (which is already predicted by the model) will be pushed to the input stream in order to detect drifts."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "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>X1</th>\n",
       "      <th>X2</th>\n",
       "      <th>X3</th>\n",
       "      <th>X4</th>\n",
       "      <th>class</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.460101</td>\n",
       "      <td>0.592744</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.588788</td>\n",
       "      <td>0.574984</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.401641</td>\n",
       "      <td>0.679325</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.306076</td>\n",
       "      <td>0.182108</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.962847</td>\n",
       "      <td>0.579245</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "    X1   X2        X3        X4  class\n",
       "0  0.0  1.0  0.460101  0.592744    1.0\n",
       "1  1.0  1.0  0.588788  0.574984    0.0\n",
       "2  0.0  0.0  0.401641  0.679325    1.0\n",
       "3  1.0  1.0  0.306076  0.182108    0.0\n",
       "4  0.0  0.0  0.962847  0.579245    1.0"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import pandas as pd\n",
    "data_path = 'https://s3.wasabisys.com/iguazio/data/function-marketplace-data/concept_drift/mixed_0101_abrupto.csv'\n",
    "predicted_train_path = 'https://s3.wasabisys.com/iguazio/data/function-marketplace-data/concept_drift/predicted_abrupto_train.csv'\n",
    "predicted_test_data_path = 'https://s3.wasabisys.com/iguazio/data/function-marketplace-data/concept_drift/predicted_abrupto_test.csv'\n",
    "# You can find the model used here\n",
    "models_path = 'https://s3.wasabisys.com/iguazio/models/function-marketplace-models/concept_drift/concept_drift_random_forest.pkl'\n",
    "original_data = pd.read_csv(data_path)\n",
    "original_data.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "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>X1</th>\n",
       "      <th>X2</th>\n",
       "      <th>X3</th>\n",
       "      <th>X4</th>\n",
       "      <th>class</th>\n",
       "      <th>predicted_col</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>34995</th>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.010106</td>\n",
       "      <td>0.647269</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>34996</th>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.293651</td>\n",
       "      <td>0.737291</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>34997</th>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.848546</td>\n",
       "      <td>0.552337</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>34998</th>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.614754</td>\n",
       "      <td>0.859896</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>34999</th>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.265306</td>\n",
       "      <td>0.843716</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        X1   X2        X3        X4  class  predicted_col\n",
       "34995  0.0  0.0  0.010106  0.647269    0.0            1.0\n",
       "34996  1.0  1.0  0.293651  0.737291    1.0            0.0\n",
       "34997  0.0  0.0  0.848546  0.552337    0.0            1.0\n",
       "34998  1.0  1.0  0.614754  0.859896    1.0            0.0\n",
       "34999  1.0  0.0  0.265306  0.843716    0.0            1.0"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "predicted_test = pd.read_csv(predicted_test_data_path)\n",
    "predicted_test.tail()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **Creating the input stream**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os \n",
    "\n",
    "container = os.path.join('/',os.environ['V3IO_HOME'].split('/')[0])\n",
    "user = os.environ[\"V3IO_USERNAME\"]\n",
    "rel_path = os.getcwd()[6:] + '/artifacts'\n",
    "\n",
    "base_input_stream = os.path.join(user,rel_path) + \"/inputs_stream\"\n",
    "base_output_stream = os.path.join(user,rel_path) + \"/output_stream\"\n",
    "input_stream = os.path.join(container,base_input_stream)\n",
    "output_stream = os.path.join(container,user,rel_path) + \"/output_stream\"\n",
    "tsdb_path = os.path.join(container,user,rel_path) + \"/output_tsdb\"\n",
    "\n",
    "stream_consumer_group = 'cg45'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import v3io.dataplane\n",
    "\n",
    "client = v3io.dataplane.Client()\n",
    "response = client.stream.create(container = container,\n",
    "                                stream_path=base_input_stream,\n",
    "                                shard_count=1,\n",
    "                                raise_for_status = v3io.dataplane.RaiseForStatus.never)\n",
    "response.raise_for_status([409, 204])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **Importing the function**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "> 2021-10-25 10:27:04,105 [info] created and saved project function-marketplace\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<mlrun.runtimes.kubejob.KubejobRuntime at 0x7f145dd80fd0>"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Importing the function\n",
    "import mlrun\n",
    "mlrun.set_environment(project='function-marketplace')\n",
    "\n",
    "fn = mlrun.import_function(\"hub://concept_drift:development\")\n",
    "fn.apply(mlrun.auto_mount())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **Running the function remotely**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "> 2021-10-25 10:27:04,567 [info] starting run concept_drift uid=fa07c222e77d4eac86d2ce9317aaded1 DB=http://mlrun-api:8080\n",
      "> 2021-10-25 10:27:04,709 [info] Job is running in the background, pod: concept-drift-ggxgb\n",
      "> 2021-10-25 10:27:11,199 [info] Loading base dataset\n",
      "> 2021-10-25 10:27:13,227 [info] Creating models\n",
      "> 2021-10-25 10:27:13,227 [info] Streaming data to models\n",
      "> 2021-10-25 10:27:13,347 [info] Logging ready models\n",
      "> 2021-10-25 10:27:13,487 [info] Deploying Concept Drift Streaming function\n",
      "> 2021-10-25 10:27:13,490 [info] Starting remote function deploy\n",
      "2021-10-25 10:27:13  (info) Deploying function\n",
      "2021-10-25 10:27:13  (info) Building\n",
      "2021-10-25 10:27:13  (info) Staging files and preparing base images\n",
      "2021-10-25 10:27:13  (info) Building processor image\n",
      "2021-10-25 10:27:15  (info) Build complete\n",
      "2021-10-25 10:27:21  (info) Function deploy complete\n",
      "> 2021-10-25 10:27:21,797 [info] successfully deployed function: {'internal_invocation_urls': ['nuclio-function-marketplace-concept-drift-streaming.default-tenant.svc.cluster.local:8080'], 'external_invocation_urls': ['default-tenant.app.dev39.lab.iguazeng.com:31143']}\n",
      "> 2021-10-25 10:27:21,868 [info] run executed, status=completed\n",
      "final state: completed\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<style>\n",
       ".dictlist {\n",
       "  background-color: #4EC64B;\n",
       "  text-align: center;\n",
       "  margin: 4px;\n",
       "  border-radius: 3px; padding: 0px 3px 1px 3px; display: inline-block;}\n",
       ".artifact {\n",
       "  cursor: pointer;\n",
       "  background-color: #4EC64B;\n",
       "  text-align: left;\n",
       "  margin: 4px; border-radius: 3px; padding: 0px 3px 1px 3px; display: inline-block;\n",
       "}\n",
       "div.block.hidden {\n",
       "  display: none;\n",
       "}\n",
       ".clickable {\n",
       "  cursor: pointer;\n",
       "}\n",
       ".ellipsis {\n",
       "  display: inline-block;\n",
       "  max-width: 60px;\n",
       "  white-space: nowrap;\n",
       "  overflow: hidden;\n",
       "  text-overflow: ellipsis;\n",
       "}\n",
       ".master-wrapper {\n",
       "  display: flex;\n",
       "  flex-flow: row nowrap;\n",
       "  justify-content: flex-start;\n",
       "  align-items: stretch;\n",
       "}\n",
       ".master-tbl {\n",
       "  flex: 3\n",
       "}\n",
       ".master-wrapper > div {\n",
       "  margin: 4px;\n",
       "  padding: 10px;\n",
       "}\n",
       "iframe.fileview {\n",
       "  border: 0 none;\n",
       "  height: 100%;\n",
       "  width: 100%;\n",
       "  white-space: pre-wrap;\n",
       "}\n",
       ".pane-header-title {\n",
       "  width: 80%;\n",
       "  font-weight: 500;\n",
       "}\n",
       ".pane-header {\n",
       "  line-height: 1;\n",
       "  background-color: #4EC64B;\n",
       "  padding: 3px;\n",
       "}\n",
       ".pane-header .close {\n",
       "  font-size: 20px;\n",
       "  font-weight: 700;\n",
       "  float: right;\n",
       "  margin-top: -5px;\n",
       "}\n",
       ".master-wrapper .right-pane {\n",
       "  border: 1px inset silver;\n",
       "  width: 40%;\n",
       "  min-height: 300px;\n",
       "  flex: 3\n",
       "  min-width: 500px;\n",
       "}\n",
       ".master-wrapper * {\n",
       "  box-sizing: border-box;\n",
       "}\n",
       "</style><script>\n",
       "function copyToClipboard(fld) {\n",
       "    if (document.queryCommandSupported && document.queryCommandSupported('copy')) {\n",
       "        var textarea = document.createElement('textarea');\n",
       "        textarea.textContent = fld.innerHTML;\n",
       "        textarea.style.position = 'fixed';\n",
       "        document.body.appendChild(textarea);\n",
       "        textarea.select();\n",
       "\n",
       "        try {\n",
       "            return document.execCommand('copy'); // Security exception may be thrown by some browsers.\n",
       "        } catch (ex) {\n",
       "\n",
       "        } finally {\n",
       "            document.body.removeChild(textarea);\n",
       "        }\n",
       "    }\n",
       "}\n",
       "function expandPanel(el) {\n",
       "  const panelName = \"#\" + el.getAttribute('paneName');\n",
       "  console.log(el.title);\n",
       "\n",
       "  document.querySelector(panelName + \"-title\").innerHTML = el.title\n",
       "  iframe = document.querySelector(panelName + \"-body\");\n",
       "\n",
       "  const tblcss = `<style> body { font-family: Arial, Helvetica, sans-serif;}\n",
       "    #csv { margin-bottom: 15px; }\n",
       "    #csv table { border-collapse: collapse;}\n",
       "    #csv table td { padding: 4px 8px; border: 1px solid silver;} </style>`;\n",
       "\n",
       "  function csvToHtmlTable(str) {\n",
       "    return '<div id=\"csv\"><table><tr><td>' +  str.replace(/[\\n\\r]+$/g, '').replace(/[\\n\\r]+/g, '</td></tr><tr><td>')\n",
       "      .replace(/,/g, '</td><td>') + '</td></tr></table></div>';\n",
       "  }\n",
       "\n",
       "  function reqListener () {\n",
       "    if (el.title.endsWith(\".csv\")) {\n",
       "      iframe.setAttribute(\"srcdoc\", tblcss + csvToHtmlTable(this.responseText));\n",
       "    } else {\n",
       "      iframe.setAttribute(\"srcdoc\", this.responseText);\n",
       "    }\n",
       "    console.log(this.responseText);\n",
       "  }\n",
       "\n",
       "  const oReq = new XMLHttpRequest();\n",
       "  oReq.addEventListener(\"load\", reqListener);\n",
       "  oReq.open(\"GET\", el.title);\n",
       "  oReq.send();\n",
       "\n",
       "\n",
       "  //iframe.src = el.title;\n",
       "  const resultPane = document.querySelector(panelName + \"-pane\");\n",
       "  if (resultPane.classList.contains(\"hidden\")) {\n",
       "    resultPane.classList.remove(\"hidden\");\n",
       "  }\n",
       "}\n",
       "function closePanel(el) {\n",
       "  const panelName = \"#\" + el.getAttribute('paneName')\n",
       "  const resultPane = document.querySelector(panelName + \"-pane\");\n",
       "  if (!resultPane.classList.contains(\"hidden\")) {\n",
       "    resultPane.classList.add(\"hidden\");\n",
       "  }\n",
       "}\n",
       "\n",
       "</script>\n",
       "<div class=\"master-wrapper\">\n",
       "  <div class=\"block master-tbl\"><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>project</th>\n",
       "      <th>uid</th>\n",
       "      <th>iter</th>\n",
       "      <th>start</th>\n",
       "      <th>state</th>\n",
       "      <th>name</th>\n",
       "      <th>labels</th>\n",
       "      <th>inputs</th>\n",
       "      <th>parameters</th>\n",
       "      <th>results</th>\n",
       "      <th>artifacts</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <td>function-marketplace</td>\n",
       "      <td><div title=\"fa07c222e77d4eac86d2ce9317aaded1\"><a href=\"https://dashboard.default-tenant.app.dev39.lab.iguazeng.com/mlprojects/function-marketplace/jobs/monitor/fa07c222e77d4eac86d2ce9317aaded1/overview\" target=\"_blank\" >...17aaded1</a></div></td>\n",
       "      <td>0</td>\n",
       "      <td>Oct 25 10:27:10</td>\n",
       "      <td>completed</td>\n",
       "      <td>concept_drift</td>\n",
       "      <td><div class=\"dictlist\">v3io_user=dani</div><div class=\"dictlist\">kind=job</div><div class=\"dictlist\">owner=dani</div><div class=\"dictlist\">host=concept-drift-ggxgb</div></td>\n",
       "      <td><div title=\"https://s3.wasabisys.com/iguazio/data/function-marketplace-data/concept_drift/predicted_abrupto_train.csv\">base_dataset</div></td>\n",
       "      <td><div class=\"dictlist\">input_stream=/users/dani/test/functions/concept_drift/artifacts/inputs_stream</div><div class=\"dictlist\">consumer_group=cg45</div><div class=\"dictlist\">output_stream=/users/dani/test/functions/concept_drift/artifacts/output_stream</div><div class=\"dictlist\">output_tsdb=/users/dani/test/functions/concept_drift/artifacts/output_tsdb</div><div class=\"dictlist\">tsdb_batch_size=1</div><div class=\"dictlist\">models=['ddm', 'eddm', 'pagehinkley']</div><div class=\"dictlist\">label_col=class</div><div class=\"dictlist\">prediction_col=predicted_col</div><div class=\"dictlist\">fn_tag=development</div></td>\n",
       "      <td></td>\n",
       "      <td><div title=\"/User/test/functions/concept_drift/artifacts/models\">eddm_concept_drift</div><div title=\"/User/test/functions/concept_drift/artifacts/models\">pagehinkley_concept_drift</div><div title=\"/User/test/functions/concept_drift/artifacts/models\">ddm_concept_drift</div></td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div></div>\n",
       "  <div id=\"resultdead8c4e-pane\" class=\"right-pane block hidden\">\n",
       "    <div class=\"pane-header\">\n",
       "      <span id=\"resultdead8c4e-title\" class=\"pane-header-title\">Title</span>\n",
       "      <span onclick=\"closePanel(this)\" paneName=\"resultdead8c4e\" class=\"close clickable\">&times;</span>\n",
       "    </div>\n",
       "    <iframe class=\"fileview\" id=\"resultdead8c4e-body\"></iframe>\n",
       "  </div>\n",
       "</div>\n"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<b> > to track results use the .show() or .logs() methods  or <a href=\"https://dashboard.default-tenant.app.dev39.lab.iguazeng.com/mlprojects/function-marketplace/jobs/monitor/fa07c222e77d4eac86d2ce9317aaded1/overview\" target=\"_blank\">click here</a> to open in UI</b>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "> 2021-10-25 10:27:23,031 [info] run executed, status=completed\n"
     ]
    }
   ],
   "source": [
    "drift_run = fn.run(name='concept_drift',\n",
    "                   params={'input_stream'    : input_stream,\n",
    "                           'consumer_group'  : stream_consumer_group,\n",
    "                           'output_stream'   : output_stream,\n",
    "                           'output_tsdb'     : tsdb_path,\n",
    "                           'tsdb_batch_size' : 1,\n",
    "                           'models'          : ['ddm', 'eddm', 'pagehinkley'], # defaults\n",
    "                           'label_col'       : 'class',\n",
    "                           'prediction_col'  : 'predicted_col',\n",
    "                           'fn_tag'          : 'development'},\n",
    "                   inputs={'base_dataset'    : predicted_train_path},\n",
    "                   artifact_path = os.path.join(os.getcwd(), 'artifacts'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **Testing the function**\n",
    "> Mark that we are testing the deployed function - concept_drift_streaming"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'data': '{\"class\": 1.0, \"request\": {\"instances\": [{\"X1\": 0.0, \"X2\": 0.0, \"X3\": 0.0634475073, \"X4\": 0.4136568818}]}, \"resp\": [1], \"when\": \"2021-10-25 10:27:23.152584\", \"model\": \"sklearn.ensemble.RandomForestClassifier\"}'}"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import json\n",
    "import datetime\n",
    "\n",
    "# Reshaping the data to V3IOStream format.\n",
    "def restructure_stream_event(context, event):\n",
    "    instances = [dict()]\n",
    "    for key in predicted_test.keys():\n",
    "        if key not in ['when', 'class', 'model', 'worker', 'hostname', 'predicted_col']:\n",
    "            instances[0].update({key: event.pop(key)})\n",
    "    event['request'] = {'instances': instances}\n",
    "    event['resp'] = [int(event.pop('predicted_col'))]\n",
    "    event['when'] = datetime.datetime.strftime(datetime.datetime.now(), format=\"%Y-%m-%d %H:%M:%S.%f\")\n",
    "    event['model'] = 'sklearn.ensemble.RandomForestClassifier'\n",
    "    return event\n",
    "    \n",
    "    \n",
    "records = json.loads(predicted_test.to_json(orient='records'))\n",
    "records = [{'data': json.dumps(restructure_stream_event(context, record))} for record in records]\n",
    "\n",
    "# showing first record\n",
    "records[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creating v3io client\n",
    "v3io_client = v3io.dataplane.Client()\n",
    "\n",
    "# Pushing some undrifted data to the input stream\n",
    "response = v3io_client.stream.put_records(container=container,\n",
    "                                          stream_path=base_input_stream, \n",
    "                                          records=records[4900:5100])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'SequenceNumber': 200,\n",
       " 'Data': 'eyJjbGFzcyI6IDAuMCwgInJlcXVlc3QiOiB7Imluc3RhbmNlcyI6IFt7IlgxIjogMC4wLCAiWDIiOiAwLjAsICJYMyI6IDAuMzMzMTYzNjk4OSwgIlg0IjogMC40MjE2NzY1Njg3fV19LCAicmVzcCI6IFsxXSwgIndoZW4iOiAiMjAyMS0xMC0yNSAxMDoyNzoyMy4yOTM3OTgiLCAibW9kZWwiOiAic2tsZWFybi5lbnNlbWJsZS5SYW5kb21Gb3Jlc3RDbGFzc2lmaWVyIn0=',\n",
       " 'ArrivalTimeSec': 1635157644,\n",
       " 'ArrivalTimeNSec': 395309631}"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Getting earliest location in the shard\n",
    "location = json.loads(v3io_client.stream.seek(container=container,\n",
    "                                              stream_path=base_input_stream,\n",
    "                                              shard_id=0,\n",
    "                                              seek_type='EARLIEST').body)['Location']\n",
    "# Getting records from input stream\n",
    "response = v3io_client.stream.get_records(container=container,\n",
    "                                          stream_path=base_input_stream,\n",
    "                                          shard_id=0, location=location)\n",
    "# Showing the last sequence that is written to the input stream\n",
    "json.loads(response.body)['Records'][-1]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Make sure some time has passed - the function needs to be triggered by the input stream, then it'll write to the output stream"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Getting earliest location in the shard\n",
    "location = json.loads(v3io_client.stream.seek(container=container,\n",
    "                                              stream_path=base_output_stream,\n",
    "                                              shard_id=0,\n",
    "                                              seek_type='EARLIEST').body)['Location']\n",
    "# Getting records from output stream\n",
    "response = v3io_client.stream.get_records(container=container,\n",
    "                                          stream_path=base_output_stream,\n",
    "                                          shard_id=0, location=location)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "sequence number : 106, data : {'class': 0.0, 'request': {'instances': [{'X1': 0.0, 'X2': 0.0, 'X3': 0.9628473804, 'X4': 0.5792453402}]}, 'resp': [1], 'when': '2021-10-25 10:27:23.291145', 'model': 'sklearn.ensemble.RandomForestClassifier', 'ddm_warning_zone': 0, 'ddm_drift': 1, 'eddm_warning_zone': 0, 'eddm_drift': 0}\n",
      "sequence number : 122, data : {'class': 0.0, 'request': {'instances': [{'X1': 0.0, 'X2': 0.0, 'X3': 0.4969765505, 'X4': 0.9784738351}]}, 'resp': [1], 'when': '2021-10-25 10:27:23.291558', 'model': 'sklearn.ensemble.RandomForestClassifier', 'ddm_warning_zone': 0, 'ddm_drift': 0, 'eddm_warning_zone': 0, 'eddm_drift': 1}\n"
     ]
    }
   ],
   "source": [
    "# Showing changed detected\n",
    "import base64\n",
    "for instance in json.loads(response.body)['Records']:\n",
    "    seq = instance[\"SequenceNumber\"]\n",
    "    data = json.loads(base64.b64decode(instance['Data']))\n",
    "    if(data['ddm_drift']==1 or data['eddm_drift']==1):\n",
    "        print(f'sequence number : {seq}, data : {data}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can see that the system detected a change in the 106 instance, which is 10006 instance in the real dataset - <br>\n",
    "5000 first instances are for train, we started pushing data from the 4900 instance of the test dataset (9900 from the real dataset), and we pushed only 200 instances.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[Back to the top](#Concept-Drift---Deployer)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python [conda env:root] *",
   "language": "python",
   "name": "conda-root-py"
  },
  "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.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}