k-means_network-anomaly.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Intrusion Detection based Anomaly method using K-Means clustering algorithm"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Importing Packages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "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": 3,
   "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": 4,
   "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": 4,
     "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": [
    "## Predict on Stream"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 1. Load Dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "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": 6,
   "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 for Training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import VectorAssembler, StringIndexer\n",
    "from pyspark.ml import Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "#train, test = data.randomSplit([0.70, 0.30])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.1 Data Vectorization and String Indexing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "label_indexer = StringIndexer(inputCol = \"status\", outputCol = \"label\")\n",
    "\n",
    "label_indexer_model = label_indexer.fit(iris)\n",
    "new_df = label_indexer_model.transform(iris)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "feature_cols = iris.columns[:-1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "assembler = VectorAssembler(inputCols = feature_cols, outputCol = 'vec_features')\n",
    "assembler_df = assembler.transform(new_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "normal = assembler_df.where(funcs.col(\"status\") == \"normal\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.2 Feature Reduction using PCA (Principal Component Analysis)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import PCA\n",
    "\n",
    "pca = PCA(k=9, inputCol=\"vec_features\", outputCol=\"features\")\n",
    "pcaModel = pca.fit(normal)\n",
    "normal_reduction_df = pcaModel.transform(normal)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "#normal_reduction_df.toPandas().head(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 3. Train Model (K-Means Clustering)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.1 Training of Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.clustering import KMeans\n",
    "\n",
    "k_num = 2\n",
    "kmeans = KMeans(featuresCol='features',k=k_num, maxIter=100)\n",
    "model = kmeans.fit(normal_reduction_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "With K=  2\n",
      "Within Set Sum of Squared Errors = 3989379345491704.0\n",
      "------------------------------------------------------------\n"
     ]
    }
   ],
   "source": [
    "wssse = model.computeCost(normal_reduction_df)\n",
    "print(\"With K= \",k_num)\n",
    "print(\"Within Set Sum of Squared Errors = \" + str(wssse))\n",
    "print('--'*30)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.1.1 Prediction Training Dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "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>features</th>\n",
       "      <th>label</th>\n",
       "      <th>prediction</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[-490.9999645758216, -0.12919664884781498, 0.1...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>[-145.9999684495366, -0.03840430885426629, 0.4...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>[4.088872553689583e-05, 3.012799773907827e-07,...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>[-234.14534818723268, 8152.938638227429, -1.34...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>[-199.11045038797118, 419.94760229094754, -0.8...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                            features  label  prediction\n",
       "0  [-490.9999645758216, -0.12919664884781498, 0.1...    0.0           0\n",
       "1  [-145.9999684495366, -0.03840430885426629, 0.4...    0.0           0\n",
       "2  [4.088872553689583e-05, 3.012799773907827e-07,...    1.0           0\n",
       "3  [-234.14534818723268, 8152.938638227429, -1.34...    0.0           0\n",
       "4  [-199.11045038797118, 419.94760229094754, -0.8...    0.0           0"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pca = PCA(k=9, inputCol=\"vec_features\", outputCol=\"features\")\n",
    "pcaModel = pca.fit(assembler_df)\n",
    "test_reduction_df = pcaModel.transform(assembler_df)\n",
    "\n",
    "predictions = model.transform(test_reduction_df)\n",
    "predictions = predictions.select(\"features\",\"label\",\"prediction\")\n",
    "predictions.toPandas().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.1.2 Calculation of Silhouette Score"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Silhouette with squared euclidean distance = 0.9999063686630117\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.evaluation import ClusteringEvaluator\n",
    "evaluator = ClusteringEvaluator()\n",
    "silhouette = evaluator.evaluate(predictions)\n",
    "print(\"Silhouette with squared euclidean distance = \" + str(silhouette))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.2 Analysing of Trained model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Analysing of how many labels have in a cluster ?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "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",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>67342</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>58624</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1</td>\n",
       "      <td>1.0</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   prediction  label  count\n",
       "0           0    0.0  67342\n",
       "1           0    1.0  58624\n",
       "2           1    0.0      1\n",
       "3           1    1.0      6"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "predictions.select(\"prediction\", \"label\").groupBy(\"prediction\", \"label\").count()\\\n",
    ".orderBy(\"prediction\", \"label\", ascending=True).toPandas().head(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "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</td>\n",
       "      <td>0.0</td>\n",
       "      <td>67342</td>\n",
       "      <td>Normal</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>58624</td>\n",
       "      <td>Anomaly</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1</td>\n",
       "      <td>Normal</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1</td>\n",
       "      <td>1.0</td>\n",
       "      <td>6</td>\n",
       "      <td>Anomaly</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   prediction  label  count   status\n",
       "0           0    0.0  67342   Normal\n",
       "1           0    1.0  58624  Anomaly\n",
       "2           1    0.0      1   Normal\n",
       "3           1    1.0      6  Anomaly"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "predictions.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": "markdown",
   "metadata": {},
   "source": [
    "### 3.3 Calculation of centroids for every cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[array([-1.20423278e+04,  3.48623574e+03,  1.60883673e+02,  1.94950168e+02,\n",
       "        -1.24698823e+02,  8.15386326e+01,  1.56155160e+00, -2.08317658e+00,\n",
       "        -1.04482667e+00]),\n",
       " array([-8.98448834e+07,  1.46547928e+06, -9.88035469e+02,  6.53455066e+02,\n",
       "        -7.73814664e+02,  1.75194282e+02, -1.31331999e+02, -3.31423494e+00,\n",
       "        -5.18382902e+00])]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_clusters = model.clusterCenters()\n",
    "\n",
    "traind_clusters = {int(i):[float(train_clusters[i][j]) for j in range(len(train_clusters[i]))] \n",
    "              for i in range(len(train_clusters))}\n",
    "train_clusters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "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>center</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   prediction                                             center\n",
       "0           0  [-12042.327840369986, 3486.2357374678886, 160....\n",
       "1           1  [-89844883.37984084, 1465479.2835174028, -988...."
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_df_centers = spark.sparkContext.parallelize([(k,)+(v,) for k,v in traind_clusters.items()]).toDF(['prediction','center'])\n",
    "train_df_centers.toPandas().head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "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>prediction</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[-490.9999645758216, -0.12919664884781498, 0.1...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>[-145.9999684495366, -0.03840430885426629, 0.4...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>[4.088872553689583e-05, 3.012799773907827e-07,...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>[-234.14534818723268, 8152.938638227429, -1.34...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>[-199.11045038797118, 419.94760229094754, -0.8...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                            features  label  prediction\n",
       "0  [-490.9999645758216, -0.12919664884781498, 0.1...    0.0           0\n",
       "1  [-145.9999684495366, -0.03840430885426629, 0.4...    0.0           0\n",
       "2  [4.088872553689583e-05, 3.012799773907827e-07,...    1.0           0\n",
       "3  [-234.14534818723268, 8152.938638227429, -1.34...    0.0           0\n",
       "4  [-199.11045038797118, 419.94760229094754, -0.8...    0.0           0"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_pred_df = predictions.withColumn('prediction',funcs.col('prediction').cast(IntegerType()))\n",
    "train_pred_df.toPandas().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.3.1 Joining of centroid and feature dataframes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "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>features</th>\n",
       "      <th>label</th>\n",
       "      <th>center</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>[-490.9999645758216, -0.12919664884781498, 0.1...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>0</td>\n",
       "      <td>[-145.9999684495366, -0.03840430885426629, 0.4...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>0</td>\n",
       "      <td>[4.088872553689583e-05, 3.012799773907827e-07,...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>0</td>\n",
       "      <td>[-234.14534818723268, 8152.938638227429, -1.34...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>0</td>\n",
       "      <td>[-199.11045038797118, 419.94760229094754, -0.8...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   prediction                                           features  label  \\\n",
       "0           0  [-490.9999645758216, -0.12919664884781498, 0.1...    0.0   \n",
       "1           0  [-145.9999684495366, -0.03840430885426629, 0.4...    0.0   \n",
       "2           0  [4.088872553689583e-05, 3.012799773907827e-07,...    1.0   \n",
       "3           0  [-234.14534818723268, 8152.938638227429, -1.34...    0.0   \n",
       "4           0  [-199.11045038797118, 419.94760229094754, -0.8...    0.0   \n",
       "\n",
       "                                              center  \n",
       "0  [-12042.327840369986, 3486.2357374678886, 160....  \n",
       "1  [-12042.327840369986, 3486.2357374678886, 160....  \n",
       "2  [-12042.327840369986, 3486.2357374678886, 160....  \n",
       "3  [-12042.327840369986, 3486.2357374678886, 160....  \n",
       "4  [-12042.327840369986, 3486.2357374678886, 160....  "
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_pred_df = train_pred_df.join(train_df_centers,on='prediction',how='left')\n",
    "train_pred_df.toPandas().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.4 Finding Anomaly Values"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Getting distance values function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "get_dist = funcs.udf(lambda features, center :\n",
    "                 float(features.squared_distance(center)),FloatType())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Sorting the furthest distance values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "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>prediction</th>\n",
       "      <th>features</th>\n",
       "      <th>label</th>\n",
       "      <th>center</th>\n",
       "      <th>dist</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>62229</th>\n",
       "      <td>0</td>\n",
       "      <td>[-344699.28443941544, 1309937355.56201, -18580...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "      <td>1.715927e+18</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>125972</th>\n",
       "      <td>1</td>\n",
       "      <td>[-1379963840.675653, -363125.98013465933, -723...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>1.664410e+18</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>125968</th>\n",
       "      <td>1</td>\n",
       "      <td>[-1167519457.088633, -307222.9200287654, -2060...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>1.161386e+18</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>125969</th>\n",
       "      <td>1</td>\n",
       "      <td>[-693375615.6526163, -182456.42189769473, -217...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>3.642521e+17</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>125967</th>\n",
       "      <td>1</td>\n",
       "      <td>[-621568642.3748622, -163560.10928325454, 1875...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>2.827328e+17</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>95024</th>\n",
       "      <td>0</td>\n",
       "      <td>[-105334.45555511035, 400291046.93912166, 3082...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "      <td>1.602301e+17</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>71625</th>\n",
       "      <td>0</td>\n",
       "      <td>[-105334.45746944736, 400291046.9404984, 30890...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "      <td>1.602301e+17</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>125966</th>\n",
       "      <td>1</td>\n",
       "      <td>[-381709077.7164838, -100442.98113029718, 2370...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>8.518716e+16</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>125970</th>\n",
       "      <td>1</td>\n",
       "      <td>[-217277332.47841534, -57174.04977907944, 2848...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>1.624135e+16</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>49320</th>\n",
       "      <td>0</td>\n",
       "      <td>[-24418806.23856078, 111734.38670165755, -665....</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "      <td>5.957018e+14</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        prediction                                           features  label  \\\n",
       "62229            0  [-344699.28443941544, 1309937355.56201, -18580...    1.0   \n",
       "125972           1  [-1379963840.675653, -363125.98013465933, -723...    1.0   \n",
       "125968           1  [-1167519457.088633, -307222.9200287654, -2060...    1.0   \n",
       "125969           1  [-693375615.6526163, -182456.42189769473, -217...    1.0   \n",
       "125967           1  [-621568642.3748622, -163560.10928325454, 1875...    1.0   \n",
       "95024            0  [-105334.45555511035, 400291046.93912166, 3082...    1.0   \n",
       "71625            0  [-105334.45746944736, 400291046.9404984, 30890...    1.0   \n",
       "125966           1  [-381709077.7164838, -100442.98113029718, 2370...    1.0   \n",
       "125970           1  [-217277332.47841534, -57174.04977907944, 2848...    1.0   \n",
       "49320            0  [-24418806.23856078, 111734.38670165755, -665....    0.0   \n",
       "\n",
       "                                                   center          dist  \n",
       "62229   [-12042.327840369986, 3486.2357374678886, 160....  1.715927e+18  \n",
       "125972  [-89844883.37984084, 1465479.2835174028, -988....  1.664410e+18  \n",
       "125968  [-89844883.37984084, 1465479.2835174028, -988....  1.161386e+18  \n",
       "125969  [-89844883.37984084, 1465479.2835174028, -988....  3.642521e+17  \n",
       "125967  [-89844883.37984084, 1465479.2835174028, -988....  2.827328e+17  \n",
       "95024   [-12042.327840369986, 3486.2357374678886, 160....  1.602301e+17  \n",
       "71625   [-12042.327840369986, 3486.2357374678886, 160....  1.602301e+17  \n",
       "125966  [-89844883.37984084, 1465479.2835174028, -988....  8.518716e+16  \n",
       "125970  [-89844883.37984084, 1465479.2835174028, -988....  1.624135e+16  \n",
       "49320   [-12042.327840369986, 3486.2357374678886, 160....  5.957018e+14  "
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_pred_df = train_pred_df.withColumn('dist',get_dist(funcs.col('features'),funcs.col('center')))\n",
    "train_pred_df.toPandas().sort_values(by=\"dist\",ascending=False).head(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.5 Calculation of Threshold Value according to distance"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.5.1 Average distance for every cluster "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "scrolled": false
   },
   "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>avgDist</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1</td>\n",
       "      <td>3.075557e+13</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>0</td>\n",
       "      <td>5.924137e+10</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   prediction       avgDist\n",
       "0           1  3.075557e+13\n",
       "1           0  5.924137e+10"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "averageDistance = train_pred_df.filter(funcs.col(\"label\") == 0.0).groupBy(\"prediction\")\\\n",
    ".agg(funcs.avg(\"dist\").alias(\"avgDist\"))\n",
    "averageDistance.toPandas().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.5.2 Maximum distance for every cluster "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "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>maxDist</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1</td>\n",
       "      <td>3.075557e+13</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>0</td>\n",
       "      <td>5.957018e+14</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   prediction       maxDist\n",
       "0           1  3.075557e+13\n",
       "1           0  5.957018e+14"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "maxDistance = train_pred_df.filter(funcs.col(\"label\") == 0.0).groupBy(\"prediction\")\\\n",
    ".agg(funcs.max(\"dist\").alias(\"maxDist\"))\n",
    "maxDistance.toPandas().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Predicted Dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "scrolled": false
   },
   "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>features</th>\n",
       "      <th>label</th>\n",
       "      <th>center</th>\n",
       "      <th>dist</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>[-490.9999645758216, -0.12919664884781498, 0.1...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "      <td>145752576.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>0</td>\n",
       "      <td>[-145.9999684495366, -0.03840430885426629, 0.4...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "      <td>153923984.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>0</td>\n",
       "      <td>[4.088872553689583e-05, 3.012799773907827e-07,...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "      <td>157467920.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>0</td>\n",
       "      <td>[-234.14534818723268, 8152.938638227429, -1.34...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "      <td>161380816.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>0</td>\n",
       "      <td>[-199.11045038797118, 419.94760229094754, -0.8...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "      <td>149992128.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   prediction                                           features  label  \\\n",
       "0           0  [-490.9999645758216, -0.12919664884781498, 0.1...    0.0   \n",
       "1           0  [-145.9999684495366, -0.03840430885426629, 0.4...    0.0   \n",
       "2           0  [4.088872553689583e-05, 3.012799773907827e-07,...    1.0   \n",
       "3           0  [-234.14534818723268, 8152.938638227429, -1.34...    0.0   \n",
       "4           0  [-199.11045038797118, 419.94760229094754, -0.8...    0.0   \n",
       "\n",
       "                                              center         dist  \n",
       "0  [-12042.327840369986, 3486.2357374678886, 160....  145752576.0  \n",
       "1  [-12042.327840369986, 3486.2357374678886, 160....  153923984.0  \n",
       "2  [-12042.327840369986, 3486.2357374678886, 160....  157467920.0  \n",
       "3  [-12042.327840369986, 3486.2357374678886, 160....  161380816.0  \n",
       "4  [-12042.327840369986, 3486.2357374678886, 160....  149992128.0  "
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_pred_df.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.5.3 Joining of predicted and threshold dataframes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "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>prediction</th>\n",
       "      <th>features</th>\n",
       "      <th>label</th>\n",
       "      <th>center</th>\n",
       "      <th>dist</th>\n",
       "      <th>prediction</th>\n",
       "      <th>maxDist</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1</td>\n",
       "      <td>[-381709077.7164838, -100442.98113029718, 2370...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>8.518716e+16</td>\n",
       "      <td>1</td>\n",
       "      <td>3.075557e+13</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>[-621568642.3748622, -163560.10928325454, 1875...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>2.827328e+17</td>\n",
       "      <td>1</td>\n",
       "      <td>3.075557e+13</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1</td>\n",
       "      <td>[-1167519457.088633, -307222.9200287654, -2060...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>1.161386e+18</td>\n",
       "      <td>1</td>\n",
       "      <td>3.075557e+13</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1</td>\n",
       "      <td>[-693375615.6526163, -182456.42189769473, -217...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>3.642521e+17</td>\n",
       "      <td>1</td>\n",
       "      <td>3.075557e+13</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>1</td>\n",
       "      <td>[-217277332.47841534, -57174.04977907944, 2848...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>1.624135e+16</td>\n",
       "      <td>1</td>\n",
       "      <td>3.075557e+13</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   prediction                                           features  label  \\\n",
       "0           1  [-381709077.7164838, -100442.98113029718, 2370...    1.0   \n",
       "1           1  [-621568642.3748622, -163560.10928325454, 1875...    1.0   \n",
       "2           1  [-1167519457.088633, -307222.9200287654, -2060...    1.0   \n",
       "3           1  [-693375615.6526163, -182456.42189769473, -217...    1.0   \n",
       "4           1  [-217277332.47841534, -57174.04977907944, 2848...    1.0   \n",
       "\n",
       "                                              center          dist  \\\n",
       "0  [-89844883.37984084, 1465479.2835174028, -988....  8.518716e+16   \n",
       "1  [-89844883.37984084, 1465479.2835174028, -988....  2.827328e+17   \n",
       "2  [-89844883.37984084, 1465479.2835174028, -988....  1.161386e+18   \n",
       "3  [-89844883.37984084, 1465479.2835174028, -988....  3.642521e+17   \n",
       "4  [-89844883.37984084, 1465479.2835174028, -988....  1.624135e+16   \n",
       "\n",
       "   prediction       maxDist  \n",
       "0           1  3.075557e+13  \n",
       "1           1  3.075557e+13  \n",
       "2           1  3.075557e+13  \n",
       "3           1  3.075557e+13  \n",
       "4           1  3.075557e+13  "
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "anomalyDetection = train_pred_df.join(maxDistance, maxDistance.prediction == train_pred_df.prediction)\n",
    "\n",
    "anomalyDetection.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.6 Assigning of labels as normal or anomaly"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "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>prediction</th>\n",
       "      <th>features</th>\n",
       "      <th>label</th>\n",
       "      <th>center</th>\n",
       "      <th>dist</th>\n",
       "      <th>prediction</th>\n",
       "      <th>maxDist</th>\n",
       "      <th>detected</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1</td>\n",
       "      <td>[-381709077.7164838, -100442.98113029718, 2370...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>8.518716e+16</td>\n",
       "      <td>1</td>\n",
       "      <td>3.075557e+13</td>\n",
       "      <td>Anomaly</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>[-621568642.3748622, -163560.10928325454, 1875...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>2.827328e+17</td>\n",
       "      <td>1</td>\n",
       "      <td>3.075557e+13</td>\n",
       "      <td>Anomaly</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1</td>\n",
       "      <td>[-1167519457.088633, -307222.9200287654, -2060...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>1.161386e+18</td>\n",
       "      <td>1</td>\n",
       "      <td>3.075557e+13</td>\n",
       "      <td>Anomaly</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1</td>\n",
       "      <td>[-693375615.6526163, -182456.42189769473, -217...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>3.642521e+17</td>\n",
       "      <td>1</td>\n",
       "      <td>3.075557e+13</td>\n",
       "      <td>Anomaly</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>1</td>\n",
       "      <td>[-217277332.47841534, -57174.04977907944, 2848...</td>\n",
       "      <td>1.0</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "      <td>1.624135e+16</td>\n",
       "      <td>1</td>\n",
       "      <td>3.075557e+13</td>\n",
       "      <td>Anomaly</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   prediction                                           features  label  \\\n",
       "0           1  [-381709077.7164838, -100442.98113029718, 2370...    1.0   \n",
       "1           1  [-621568642.3748622, -163560.10928325454, 1875...    1.0   \n",
       "2           1  [-1167519457.088633, -307222.9200287654, -2060...    1.0   \n",
       "3           1  [-693375615.6526163, -182456.42189769473, -217...    1.0   \n",
       "4           1  [-217277332.47841534, -57174.04977907944, 2848...    1.0   \n",
       "\n",
       "                                              center          dist  \\\n",
       "0  [-89844883.37984084, 1465479.2835174028, -988....  8.518716e+16   \n",
       "1  [-89844883.37984084, 1465479.2835174028, -988....  2.827328e+17   \n",
       "2  [-89844883.37984084, 1465479.2835174028, -988....  1.161386e+18   \n",
       "3  [-89844883.37984084, 1465479.2835174028, -988....  3.642521e+17   \n",
       "4  [-89844883.37984084, 1465479.2835174028, -988....  1.624135e+16   \n",
       "\n",
       "   prediction       maxDist detected  \n",
       "0           1  3.075557e+13  Anomaly  \n",
       "1           1  3.075557e+13  Anomaly  \n",
       "2           1  3.075557e+13  Anomaly  \n",
       "3           1  3.075557e+13  Anomaly  \n",
       "4           1  3.075557e+13  Anomaly  "
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "detected_df = anomalyDetection.withColumn(\"detected\", funcs.when(anomalyDetection.dist > anomalyDetection.maxDist, \"Anomaly\").otherwise(\"Normal\"))\n",
    "detected_df.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.7 Evaluation of result using Confusion Matrix"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "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>label</th>\n",
       "      <th>detected</th>\n",
       "      <th>count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Anomaly</td>\n",
       "      <td>Normal</td>\n",
       "      <td>58621</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Anomaly</td>\n",
       "      <td>Anomaly</td>\n",
       "      <td>9</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Normal</td>\n",
       "      <td>Normal</td>\n",
       "      <td>67343</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "     label detected  count\n",
       "0  Anomaly   Normal  58621\n",
       "1  Anomaly  Anomaly      9\n",
       "2   Normal   Normal  67343"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "conf_matrix = detected_df.withColumn(\"label\",funcs.when(funcs.col(\"label\").isin(1), \"Anomaly\").otherwise(\"Normal\"))\\\n",
    ".groupBy(\"label\",\"detected\").count()\n",
    "conf_matrix.toPandas().head(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.7.1 Calculation of Accuracy "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "all_df = detected_df.count()\n",
    "tptn = conf_matrix.filter(conf_matrix.label == conf_matrix.detected).agg(funcs.sum(\"count\")).select(\"sum(count)\").toPandas().head()\n",
    "tptn = tptn.at[0, 'sum(count)']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Accuracy:  0.53465425130782\n"
     ]
    }
   ],
   "source": [
    "accuracy = tptn / all_df\n",
    "print(\"Accuracy: \", accuracy)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.7.2 Calculation of Recall"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "tp = conf_matrix.filter((conf_matrix.label == \"Normal\") & ( conf_matrix.detected == \"Normal\")).select(\"count\").toPandas()\n",
    "fn = conf_matrix.filter((conf_matrix.label == \"Anomaly\") & ( conf_matrix.detected == \"Normal\")).select(\"count\").toPandas()\n",
    "tp = tp.at[0, 'count']\n",
    "fn = fn.at[0, 'count']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Recall:  0.5346210028262043\n"
     ]
    }
   ],
   "source": [
    "recall = tp / (tp + fn)\n",
    "print(\"Recall: \", recall)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 4. Streaming Process"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "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": "code",
   "execution_count": 38,
   "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": 39,
   "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": 40,
   "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": [
    "## 4.1 Vectorization of streaming data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "tovec_udf = funcs.udf(lambda r: Vectors.dense(r), VectorUDT())\n",
    "data_stream = features_array.withColumn(\"vec_features\", tovec_udf(\"arr\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4.2 Feature Reduction "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "reduced_data_stream = pcaModel.transform(data_stream)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 5. Prediction Process"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.1 Prediction of Streaming Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "streaming_prediction = model.transform(reduced_data_stream)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "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",
      " |-- vec_features: vector (nullable = true)\n",
      " |-- features: vector (nullable = true)\n",
      " |-- prediction: integer (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "streaming_prediction.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.2 Finding Anomaly"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "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>center</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>[-12042.327840369986, 3486.2357374678886, 160....</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>[-89844883.37984084, 1465479.2835174028, -988....</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   prediction                                             center\n",
       "0           0  [-12042.327840369986, 3486.2357374678886, 160....\n",
       "1           1  [-89844883.37984084, 1465479.2835174028, -988...."
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_df_centers.toPandas().head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_pred = streaming_prediction.withColumn('prediction',funcs.col('prediction').cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_pred = df_pred.join(train_df_centers,on='prediction',how='left')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.2.1 Getting Euclidean Distance for features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [],
   "source": [
    "get_dist = funcs.udf(lambda features, center : \n",
    "                 float(features.squared_distance(center)),FloatType())\n",
    "\n",
    "df_pred = df_pred.withColumn('dist',get_dist(funcs.col('features'),funcs.col('center')))\n",
    "df_pred = df_pred.withColumnRenamed(\"prediction\", \"cluster\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.2.2 Calculation of Average and Maximum Distance for Streaming"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- cluster: integer (nullable = false)\n",
      " |-- arr: array (nullable = false)\n",
      " |    |-- element: float (containsNull = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- vec_features: vector (nullable = true)\n",
      " |-- features: vector (nullable = true)\n",
      " |-- center: array (nullable = true)\n",
      " |    |-- element: double (containsNull = true)\n",
      " |-- dist: float (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_pred.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [],
   "source": [
    "maxDist_streaming = df_pred.join(maxDistance, maxDistance.prediction == df_pred.cluster)\n",
    "avgDist_streaming = df_pred.join(averageDistance, averageDistance.prediction == df_pred.cluster)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "detection = maxDist_streaming.withColumn(\"detected\", funcs.when(maxDist_streaming.dist > maxDist_streaming.maxDist, \"Anomaly\").otherwise(\"Normal\"))\n",
    "detection = detection.select(\"features\", \"prediction\", \"dist\", \"maxDist\", \"status\", \"detected\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- features: vector (nullable = true)\n",
      " |-- prediction: integer (nullable = false)\n",
      " |-- dist: float (nullable = true)\n",
      " |-- maxDist: float (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- detected: string (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "detection.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.3 Adding Sliding Window Time using Current Timestamp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [],
   "source": [
    "currentTimeDf = detection.withColumn(\"processingTime\",funcs.current_timestamp())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- features: vector (nullable = true)\n",
      " |-- prediction: integer (nullable = false)\n",
      " |-- dist: float (nullable = true)\n",
      " |-- maxDist: float (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- detected: string (nullable = false)\n",
      " |-- processingTime: timestamp (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "currentTimeDf.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "windowedCount = currentTimeDf\\\n",
    ".withWatermark(\"processingTime\", \"5 seconds\")\\\n",
    ".groupBy(funcs.window(\"processingTime\", \"4 seconds\", \"2 seconds\"), \"prediction\", \n",
    "                                      \"status\", \"detected\", \"dist\", \"maxDist\")\\\n",
    ".avg(\"dist\").orderBy(\"window\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 6. Start Streaming"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 6.1 Option 1 - Using Sliding Window and Watermarking (Confusion Matrix)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [],
   "source": [
    "confusion_matrix = detection.groupBy(\"status\",\"detected\").count().select(\"status\", \"detected\",\"count\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [],
   "source": [
    "confusion_matrix = currentTimeDf\\\n",
    ".withWatermark(\"processingTime\", \"5 seconds\")\\\n",
    ".groupBy(funcs.window(\"processingTime\", \"3 seconds\", \"1 seconds\"),\"status\", \"detected\")\\\n",
    ".count().orderBy(\"window\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "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": 63,
   "metadata": {},
   "outputs": [],
   "source": [
    "q = windowedCount.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.3 Option 3 - Using Complete and Aggregate Function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "group = df_pred.groupBy([\"cluster\",\"status\"]).count()\\\n",
    ".orderBy(\"cluster\", \"status\", ascending=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [],
   "source": [
    "q = group.writeStream\\\n",
    ".outputMode(\"complete\")\\\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
}