decision_tree_network_anomaly.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Intrusion Detection based Anomaly method using Decision Tree algorithm"
   ]
  },
  {
   "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 Decision Tree Algorithm"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "from pyspark.ml.classification import DecisionTreeClassifier"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.1.1 Training and Predicting of Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "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>(4675.0, 459.0, 81673.0, 0.0, 0.0, 0.0, 2.0, 1...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[639.0, 43.0]</td>\n",
       "      <td>[0.9369501466275659, 0.06304985337243402]</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>(583.0, 848.0, 25323.0, 0.0, 0.0, 0.0, 1.0, 0....</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[639.0, 43.0]</td>\n",
       "      <td>[0.9369501466275659, 0.06304985337243402]</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>(169.0, 1567.0, 2857.0, 0.0, 0.0, 0.0, 3.0, 0....</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[639.0, 43.0]</td>\n",
       "      <td>[0.9369501466275659, 0.06304985337243402]</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>(179.0, 1559.0, 2855.0, 0.0, 0.0, 0.0, 3.0, 0....</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[639.0, 43.0]</td>\n",
       "      <td>[0.9369501466275659, 0.06304985337243402]</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>(454.0, 492.0, 14334.0, 0.0, 0.0, 0.0, 2.0, 0....</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[639.0, 43.0]</td>\n",
       "      <td>[0.9369501466275659, 0.06304985337243402]</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                            features  label  rawPrediction  \\\n",
       "0  (4675.0, 459.0, 81673.0, 0.0, 0.0, 0.0, 2.0, 1...    0.0  [639.0, 43.0]   \n",
       "1  (583.0, 848.0, 25323.0, 0.0, 0.0, 0.0, 1.0, 0....    0.0  [639.0, 43.0]   \n",
       "2  (169.0, 1567.0, 2857.0, 0.0, 0.0, 0.0, 3.0, 0....    1.0  [639.0, 43.0]   \n",
       "3  (179.0, 1559.0, 2855.0, 0.0, 0.0, 0.0, 3.0, 0....    1.0  [639.0, 43.0]   \n",
       "4  (454.0, 492.0, 14334.0, 0.0, 0.0, 0.0, 2.0, 0....    0.0  [639.0, 43.0]   \n",
       "\n",
       "                                 probability  prediction  \n",
       "0  [0.9369501466275659, 0.06304985337243402]         0.0  \n",
       "1  [0.9369501466275659, 0.06304985337243402]         0.0  \n",
       "2  [0.9369501466275659, 0.06304985337243402]         0.0  \n",
       "3  [0.9369501466275659, 0.06304985337243402]         0.0  \n",
       "4  [0.9369501466275659, 0.06304985337243402]         0.0  "
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dt = DecisionTreeClassifier(labelCol=\"label\", featuresCol=\"features\")\n",
    "modeldt = dt.fit(train)\n",
    "predictiondt = modeldt.transform(test)\n",
    "predictiondt.toPandas().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.1.2 Confusion Matrix of Decision Tree"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "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>prediction</th>\n",
       "      <th>label</th>\n",
       "      <th>count</th>\n",
       "      <th>status</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20034</td>\n",
       "      <td>Normal</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>249</td>\n",
       "      <td>Anomaly</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>150</td>\n",
       "      <td>Normal</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>17218</td>\n",
       "      <td>Anomaly</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   prediction  label  count   status\n",
       "0         0.0    0.0  20034   Normal\n",
       "1         0.0    1.0    249  Anomaly\n",
       "2         1.0    0.0    150   Normal\n",
       "3         1.0    1.0  17218  Anomaly"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "predictiondt.select(\"prediction\", \"label\")\\\n",
    ".groupBy(\"prediction\", \"label\").count()\\\n",
    ".orderBy(\"prediction\", \"label\", ascending=True).withColumn(\"status\",\n",
    "funcs.when(funcs.col(\"label\").isin(1), \"Anomaly\")\\\n",
    ".otherwise(\"Normal\")).toPandas().head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "scrolled": true
   },
   "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>label</th>\n",
       "      <th>prediction</th>\n",
       "      <th>count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>17218</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>150</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>249</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20034</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   label  prediction  count\n",
       "0    1.0         1.0  17218\n",
       "1    0.0         1.0    150\n",
       "2    1.0         0.0    249\n",
       "3    0.0         0.0  20034"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "predictiondt.groupBy([\"label\",\"prediction\"]).count().toPandas().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.1.3 Calculation of Accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--- Decision Tree --- \n",
      "Accuracy Rate = 0.9894\n",
      "  Error  Rate = 0.0106 \n"
     ]
    }
   ],
   "source": [
    "evaluatordt = MulticlassClassificationEvaluator(\n",
    "    labelCol=\"label\", predictionCol=\"prediction\", metricName=\"accuracy\")\n",
    "dt = evaluatordt.evaluate(predictiondt)\n",
    "\n",
    "print(\"--- Decision Tree --- \")\n",
    "print(\"Accuracy Rate =\", round(dt,4))\n",
    "print(\"  Error  Rate = %g \" % round((1.0 - dt),4))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+\n",
      "| normal|anomaly|\n",
      "+-------+-------+\n",
      "|20034.0|  150.0|\n",
      "|  249.0|17218.0|\n",
      "+-------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "predictionAndLabel = predictiondt.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": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+-----+-----+\n",
      "|prediction_label|  0.0|  1.0|\n",
      "+----------------+-----+-----+\n",
      "|             1.0|  150|17218|\n",
      "|             0.0|20034|  249|\n",
      "+----------------+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "predictiondt.withColumn(\"A\", funcs.struct(\"prediction\",\"label\")).crosstab(\"prediction\",\"label\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 4. Streaming Process"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "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": 24,
   "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": 25,
   "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": 26,
   "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": 27,
   "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": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "prediction = modeldt.transform(data_stream)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(prediction)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "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": 33,
   "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": 34,
   "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": 35,
   "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": 37,
   "metadata": {},
   "outputs": [],
   "source": [
    "prediction = prediction.select(\"features\",\"status\",\"prediction\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "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
}