naive_bayes_network-anomaly.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Intrusion Detection based Anomaly method using Naive Bayes  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Importing Packages and configuring spark engine"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as funcs\n",
    "from pyspark.ml.linalg import Vectors, VectorUDT\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder\\\n",
    ".master(\"local[4]\")\\\n",
    ".appName(\"ReadFromCsv\")\\\n",
    ".config(\"spark.driver.memory\",\"3g\")\\\n",
    ".config(\"spark.executor.memory\", \"4g\")\\\n",
    ".getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'logger = spark.sparkContext._jvm.org.apache.log4j\\nlogger.LogManager.getLogger(\"org\"). setLevel(logger.Level.ERROR)\\nlogger.LogManager.getLogger(\"akka\").setLevel(logger.Level.ERROR)'"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "'''logger = spark.sparkContext._jvm.org.apache.log4j\n",
    "logger.LogManager.getLogger(\"org\"). setLevel(logger.Level.ERROR)\n",
    "logger.LogManager.getLogger(\"akka\").setLevel(logger.Level.ERROR)'''"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 1. Load Dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "iris = spark.read \\\n",
    ".format(\"csv\")\\\n",
    ".option(\"header\", True)\\\n",
    ".option(\"sep\", \",\")\\\n",
    ".option(\"inferSchema\", \"True\")\\\n",
    ".load(\"TrainDf.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- duration: integer (nullable = true)\n",
      " |-- src_bytes: integer (nullable = true)\n",
      " |-- dst_bytes: integer (nullable = true)\n",
      " |-- land: integer (nullable = true)\n",
      " |-- wrong_fragment: integer (nullable = true)\n",
      " |-- urgent: integer (nullable = true)\n",
      " |-- hot: integer (nullable = true)\n",
      " |-- num_failed_logins: integer (nullable = true)\n",
      " |-- logged_in: integer (nullable = true)\n",
      " |-- num_compromised: integer (nullable = true)\n",
      " |-- root_shell: integer (nullable = true)\n",
      " |-- su_attempted: integer (nullable = true)\n",
      " |-- num_root: integer (nullable = true)\n",
      " |-- num_file_creations: integer (nullable = true)\n",
      " |-- num_shells: integer (nullable = true)\n",
      " |-- num_access_files: integer (nullable = true)\n",
      " |-- num_outbound_cmds: integer (nullable = true)\n",
      " |-- is_host_login: integer (nullable = true)\n",
      " |-- is_guest_login: integer (nullable = true)\n",
      " |-- count: integer (nullable = true)\n",
      " |-- srv_count: integer (nullable = true)\n",
      " |-- serror_rate: double (nullable = true)\n",
      " |-- srv_serror_rate: double (nullable = true)\n",
      " |-- rerror_rate: double (nullable = true)\n",
      " |-- srv_rerror_rate: double (nullable = true)\n",
      " |-- same_srv_rate: double (nullable = true)\n",
      " |-- diff_srv_rate: double (nullable = true)\n",
      " |-- srv_diff_host_rate: double (nullable = true)\n",
      " |-- dst_host_count: integer (nullable = true)\n",
      " |-- dst_host_srv_count: integer (nullable = true)\n",
      " |-- dst_host_same_srv_rate: double (nullable = true)\n",
      " |-- dst_host_diff_srv_rate: double (nullable = true)\n",
      " |-- dst_host_same_src_port_rate: double (nullable = true)\n",
      " |-- dst_host_srv_diff_host_rate: double (nullable = true)\n",
      " |-- dst_host_serror_rate: double (nullable = true)\n",
      " |-- dst_host_srv_serror_rate: double (nullable = true)\n",
      " |-- dst_host_rerror_rate: double (nullable = true)\n",
      " |-- dst_host_srv_rerror_rate: double (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "iris.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2. Data Preparation "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import VectorAssembler, StringIndexer\n",
    "from pyspark.ml import Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "feature_cols = iris.columns[:-1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "label_indexer = StringIndexer(inputCol = \"status\", outputCol = \"label\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "assembler = VectorAssembler(inputCols = feature_cols, outputCol = 'features')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "pipe = Pipeline(stages=[assembler, label_indexer])\n",
    "pipe_model = pipe.fit(iris)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = pipe_model.transform(iris)\n",
    "data = data.select(\"features\",\"label\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "train, test = data.randomSplit([0.70, 0.30])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 3. Train Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.1 Naive Bayes Algorithm"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import NaiveBayes\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.1.1 Training and Predicting of Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "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>features</th>\n",
       "      <th>label</th>\n",
       "      <th>rawPrediction</th>\n",
       "      <th>probability</th>\n",
       "      <th>prediction</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>(2399.0, 2238.0, 460350.0, 0.0, 0.0, 3.0, 0.0,...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[-649979.971662576, -617163.2631252023]</td>\n",
       "      <td>[0.0, 1.0]</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>(7805.0, 1828.0, 148722.0, 0.0, 0.0, 0.0, 1.0,...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[-243354.11048568296, -242178.04531887968]</td>\n",
       "      <td>[0.0, 1.0]</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>(718.0, 1412.0, 25260.0, 0.0, 0.0, 0.0, 15.0, ...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-40050.94681206568, -39731.970064071844]</td>\n",
       "      <td>[2.9522864334545822e-139, 1.0]</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                            features  label  \\\n",
       "0  (2399.0, 2238.0, 460350.0, 0.0, 0.0, 3.0, 0.0,...    0.0   \n",
       "1  (7805.0, 1828.0, 148722.0, 0.0, 0.0, 0.0, 1.0,...    0.0   \n",
       "2  (718.0, 1412.0, 25260.0, 0.0, 0.0, 0.0, 15.0, ...    1.0   \n",
       "\n",
       "                                rawPrediction                     probability  \\\n",
       "0     [-649979.971662576, -617163.2631252023]                      [0.0, 1.0]   \n",
       "1  [-243354.11048568296, -242178.04531887968]                      [0.0, 1.0]   \n",
       "2   [-40050.94681206568, -39731.970064071844]  [2.9522864334545822e-139, 1.0]   \n",
       "\n",
       "   prediction  \n",
       "0         1.0  \n",
       "1         1.0  \n",
       "2         1.0  "
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "nb = NaiveBayes(smoothing=1.0, modelType=\"multinomial\")\n",
    "\n",
    "modelnb = nb.fit(train)\n",
    "predictionnb = modelnb.transform(test)\n",
    "predictionnb.toPandas().head(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.1.2 Calculation of Accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--- Naive Bayes --- \n",
      "Accuracy Rate = 0.5135\n",
      "  Error  Rate = 0.4865 \n"
     ]
    }
   ],
   "source": [
    "evaluatornb = MulticlassClassificationEvaluator(labelCol=\"label\", \n",
    "                                              predictionCol=\"prediction\", \n",
    "                                              metricName=\"accuracy\")\n",
    "nb = evaluatornb.evaluate(predictionnb)\n",
    "\n",
    "print(\"--- Naive Bayes --- \")\n",
    "print(\"Accuracy Rate =\", round(nb,4))\n",
    "print(\"  Error  Rate = %g \" % round((1.0 - nb),4))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.1.3 Confusion Matrix of Decision Tree"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+\n",
      "| normal|anomaly|\n",
      "+-------+-------+\n",
      "|19073.0| 1102.0|\n",
      "|17180.0|  223.0|\n",
      "+-------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "predictionAndLabel = predictionnb.select(\"prediction\", \"label\").rdd\n",
    "\n",
    "from pyspark.mllib.evaluation import MulticlassMetrics\n",
    "metrics = MulticlassMetrics(predictionAndLabel)\n",
    "cm = metrics.confusionMatrix()\n",
    "rows = cm.toArray().tolist()\n",
    "\n",
    "confusion_matrix = spark.createDataFrame(rows,[\"normal\",\"anomaly\"])\n",
    "confusion_matrix.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+----------+-----+\n",
      "|label|prediction|count|\n",
      "+-----+----------+-----+\n",
      "|  1.0|       1.0|  223|\n",
      "|  0.0|       1.0| 1102|\n",
      "|  1.0|       0.0|17180|\n",
      "|  0.0|       0.0|19073|\n",
      "+-----+----------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "predictionrfevaluator = MulticlassClassificationEvaluator(predictionCol=\"prediction\", \\\n",
    "                    labelCol=\"label\",metricName=\"accuracy\")\n",
    "predictionrfevaluator.evaluate(predictionnb)      \n",
    "\n",
    "predictionnb.groupBy(\"label\",\"prediction\").count().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 4. Streaming Process"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "schema = StructType(\n",
    "[\n",
    "    StructField(\"duration\", FloatType(), True),\n",
    "    StructField(\"src_bytes\", FloatType(), True),\n",
    "    StructField(\"dst_bytes\", FloatType(), True),\n",
    "    StructField(\"land\", FloatType(), True),\n",
    "    StructField(\"wrong_fragment\", FloatType(), True),\n",
    "    StructField(\"urgent\", FloatType(), True),\n",
    "    StructField(\"hot\", FloatType(), True),\n",
    "    StructField(\"num_failed_logins\", FloatType(), True),\n",
    "    StructField(\"logged_in\", FloatType(), True),\n",
    "    StructField(\"num_compromised\", FloatType(), True),\n",
    "    StructField(\"root_shell\", FloatType(), True),\n",
    "    StructField(\"su_attempted\", FloatType(), True),\n",
    "    StructField(\"num_root\", FloatType(), True),\n",
    "    StructField(\"num_file_creations\", FloatType(), True),\n",
    "    StructField(\"num_shells\", FloatType(), True),\n",
    "    StructField(\"num_access_files\", FloatType(), True),\n",
    "    StructField(\"num_outbound_cmds\", FloatType(), True),\n",
    "    StructField(\"is_host_login\", FloatType(), True),\n",
    "    StructField(\"is_guest_login\", FloatType(), True),\n",
    "    StructField(\"count\", FloatType(), True),\n",
    "    StructField(\"srv_count\", FloatType(), True),\n",
    "    StructField(\"serror_rate\", FloatType(), True),\n",
    "    StructField(\"srv_serror_rate\", FloatType(), True),\n",
    "    StructField(\"rerror_rate\", FloatType(), True),\n",
    "    StructField(\"srv_rerror_rate\", FloatType(), True),\n",
    "    StructField(\"same_srv_rate\", FloatType(), True),\n",
    "    StructField(\"diff_srv_rate\", FloatType(), True),\n",
    "    StructField(\"srv_diff_host_rate\", FloatType(), True),\n",
    "    StructField(\"dst_host_count\", FloatType(), True),\n",
    "    StructField(\"dst_host_srv_count\", FloatType(), True),\n",
    "    StructField(\"dst_host_same_srv_rate\", FloatType(), True),\n",
    "    StructField(\"dst_host_diff_srv_rate\", FloatType(), True),\n",
    "    StructField(\"dst_host_same_src_port_rate\", FloatType(), True),\n",
    "    StructField(\"dst_host_srv_diff_host_rate\", FloatType(), True),\n",
    "    StructField(\"dst_host_serror_rate\", FloatType(), True),\n",
    "    StructField(\"dst_host_srv_serror_rate\", FloatType(), True),\n",
    "    StructField(\"dst_host_rerror_rate\", FloatType(), True),\n",
    "    StructField(\"dst_host_srv_rerror_rate\", FloatType(), True),\n",
    "    \n",
    "    StructField(\"status\", StringType(), True)\n",
    "])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.1 Reading Streaming Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "iris_data = spark.readStream \\\n",
    ".format(\"csv\")\\\n",
    ".option(\"header\", True)\\\n",
    ".option(\"sep\", \",\")\\\n",
    ".schema(schema)\\\n",
    ".load(\"data\")\\"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- duration: integer (nullable = true)\n",
      " |-- src_bytes: integer (nullable = true)\n",
      " |-- dst_bytes: integer (nullable = true)\n",
      " |-- land: integer (nullable = true)\n",
      " |-- wrong_fragment: integer (nullable = true)\n",
      " |-- urgent: integer (nullable = true)\n",
      " |-- hot: integer (nullable = true)\n",
      " |-- num_failed_logins: integer (nullable = true)\n",
      " |-- logged_in: integer (nullable = true)\n",
      " |-- num_compromised: integer (nullable = true)\n",
      " |-- root_shell: integer (nullable = true)\n",
      " |-- su_attempted: integer (nullable = true)\n",
      " |-- num_root: integer (nullable = true)\n",
      " |-- num_file_creations: integer (nullable = true)\n",
      " |-- num_shells: integer (nullable = true)\n",
      " |-- num_access_files: integer (nullable = true)\n",
      " |-- num_outbound_cmds: integer (nullable = true)\n",
      " |-- is_host_login: integer (nullable = true)\n",
      " |-- is_guest_login: integer (nullable = true)\n",
      " |-- count: integer (nullable = true)\n",
      " |-- srv_count: integer (nullable = true)\n",
      " |-- serror_rate: double (nullable = true)\n",
      " |-- srv_serror_rate: double (nullable = true)\n",
      " |-- rerror_rate: double (nullable = true)\n",
      " |-- srv_rerror_rate: double (nullable = true)\n",
      " |-- same_srv_rate: double (nullable = true)\n",
      " |-- diff_srv_rate: double (nullable = true)\n",
      " |-- srv_diff_host_rate: double (nullable = true)\n",
      " |-- dst_host_count: integer (nullable = true)\n",
      " |-- dst_host_srv_count: integer (nullable = true)\n",
      " |-- dst_host_same_srv_rate: double (nullable = true)\n",
      " |-- dst_host_diff_srv_rate: double (nullable = true)\n",
      " |-- dst_host_same_src_port_rate: double (nullable = true)\n",
      " |-- dst_host_srv_diff_host_rate: double (nullable = true)\n",
      " |-- dst_host_serror_rate: double (nullable = true)\n",
      " |-- dst_host_srv_serror_rate: double (nullable = true)\n",
      " |-- dst_host_rerror_rate: double (nullable = true)\n",
      " |-- dst_host_srv_rerror_rate: double (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "iris.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "features_array = iris_data.selectExpr(\"\"\"array(\n",
    "CAST(duration AS FLOAT),\n",
    "CAST(src_bytes AS FLOAT), \n",
    "CAST(dst_bytes AS FLOAT), \n",
    "CAST(land AS FLOAT),\n",
    "CAST(wrong_fragment AS FLOAT), \n",
    "CAST(urgent AS FLOAT), \n",
    "CAST(hot AS FLOAT), \n",
    "CAST(num_failed_logins AS FLOAT), \n",
    "CAST(logged_in AS FLOAT),\n",
    "CAST(num_compromised AS FLOAT), \n",
    "CAST(root_shell AS FLOAT),\n",
    "CAST(su_attempted AS FLOAT), \n",
    "CAST(num_root AS FLOAT),\n",
    "CAST(num_file_creations AS FLOAT), \n",
    "CAST(num_shells  AS FLOAT), \n",
    "CAST(num_access_files  AS FLOAT),\n",
    "CAST(num_outbound_cmds  AS FLOAT), \n",
    "CAST(is_host_login  AS FLOAT),\n",
    "CAST(is_guest_login  AS FLOAT), \n",
    "CAST(count AS FLOAT), \n",
    "CAST(srv_count AS FLOAT),\n",
    "CAST(serror_rate AS FLOAT), \n",
    "CAST(srv_serror_rate AS FLOAT),\n",
    "CAST(rerror_rate AS FLOAT), \n",
    "CAST(srv_rerror_rate AS FLOAT),\n",
    "CAST(same_srv_rate AS FLOAT),\n",
    "CAST(diff_srv_rate AS FLOAT), \n",
    "CAST(srv_diff_host_rate AS FLOAT),\n",
    "CAST(dst_host_count AS FLOAT),\n",
    "CAST(dst_host_srv_count AS FLOAT), \n",
    "CAST(dst_host_same_srv_rate AS FLOAT),\n",
    "CAST(dst_host_diff_srv_rate AS FLOAT),\n",
    "CAST(dst_host_same_src_port_rate AS FLOAT),\n",
    "CAST(dst_host_srv_diff_host_rate AS FLOAT), \n",
    "CAST(dst_host_serror_rate AS FLOAT),\n",
    "CAST(dst_host_srv_serror_rate AS FLOAT),\n",
    "CAST(dst_host_rerror_rate AS FLOAT), \n",
    "CAST(dst_host_srv_rerror_rate AS FLOAT)\n",
    "\n",
    ") as arr\"\"\", \n",
    "                                      \"status\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.2 Vectorization of streaming data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "tovec_udf = funcs.udf(lambda r: Vectors.dense(r), VectorUDT())\n",
    "data_stream = features_array.withColumn(\"features\", tovec_udf(\"arr\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 5. Prediction Process"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.1 Prediction of Streaming Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "prediction = modelnb.transform(data_stream)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(prediction)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- arr: array (nullable = false)\n",
      " |    |-- element: float (containsNull = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- features: vector (nullable = true)\n",
      " |-- rawPrediction: vector (nullable = true)\n",
      " |-- probability: vector (nullable = true)\n",
      " |-- prediction: double (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "prediction.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.2 Adding Sliding Window Time using Current Timestamp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "currentTimeDf = prediction.withColumn(\"processingTime\",funcs.current_timestamp())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 6. Start Streaming"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 6.1 Option 1 - Using Sliding Windows and Watermarking (Confusion Matrix)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "confusion_matrix = currentTimeDf\\\n",
    ".withWatermark(\"processingTime\", \"5 seconds\")\\\n",
    ".groupBy(funcs.window(\"processingTime\", \"3 seconds\", \"1 seconds\"),\"status\", \"prediction\")\\\n",
    ".count()\\\n",
    ".withColumn(\"prediction\",funcs.when(funcs.col(\"prediction\").isin(1.0), \"anomaly\")\\\n",
    ".otherwise(\"normal\"))\\\n",
    ".orderBy(\"window\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "q = confusion_matrix.writeStream\\\n",
    ".outputMode(\"complete\")\\\n",
    ".format(\"console\")\\\n",
    ".option(\"truncate\", \"false\")\\\n",
    ".start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q.awaitTermination()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 6.2 Option 2 - Using Append method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "prediction = prediction.select(\"features\",\"status\",\"prediction\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "q = prediction.writeStream\\\n",
    ".outputMode(\"append\")\\\n",
    ".format(\"console\")\\\n",
    ".start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q.awaitTermination()"
   ]
  },
  {
   "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.7.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}