Course_Project_Option_2.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2> Course Project with IOT dataset </h2>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "import json\n",
    "findspark.init()\n",
    "findspark.find()\n",
    "\n",
    "\n",
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark import SparkContext, SQLContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "C:\\spark\\python\\pyspark\\sql\\context.py:113: FutureWarning: Deprecated in 3.0.0. Use SparkSession.builder.getOrCreate() instead.\n",
      "  warnings.warn(\n"
     ]
    }
   ],
   "source": [
    "appName = \"Big Data Analytics\"\n",
    "master = \"local\"\n",
    "\n",
    "# Create Configuration object for Spark.\n",
    "conf = pyspark.SparkConf()\\\n",
    "    .set('spark.driver.host','127.0.0.1')\\\n",
    "    .setAppName(appName)\\\n",
    "    .setMaster(master)\n",
    "\n",
    "# Create Spark Context with the new configurations rather than relying on the default one\n",
    "sc = SparkContext.getOrCreate(conf=conf)\n",
    "\n",
    "# You need to create SQL Context to conduct some database operations like what we will see later.\n",
    "sqlContext = SQLContext(sc)\n",
    "\n",
    "# If you have SQL context, you create the session from the Spark Context\n",
    "# spark = sqlContext.sparkSession.builder.getOrCreate()\n",
    "spark = sqlContext.sparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_Train = spark.read.csv(\"train70_reduced.csv\",header=True, inferSchema= True)\n",
    "df_Test = spark.read.csv(\"test30_reduced.csv\",header=True, inferSchema= True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Added a new column to both datasets to distunguish between the two\n",
    "import pyspark.sql.functions as pyspark_funcs\n",
    "df_Train = df_Train.withColumn('is_train', pyspark_funcs.lit(1))\n",
    "df_Test = df_Test.withColumn('is_train', pyspark_funcs.lit(0))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-RECORD 0--------------------------------\n",
      " tcp.flags                  | 0x00000018 \n",
      " tcp.time_delta             | 0.998867   \n",
      " tcp.len                    | 10         \n",
      " mqtt.conack.flags          | 0          \n",
      " mqtt.conack.flags.reserved | 0.0        \n",
      " mqtt.conack.flags.sp       | 0.0        \n",
      " mqtt.conack.val            | 0.0        \n",
      " mqtt.conflag.cleansess     | 0.0        \n",
      " mqtt.conflag.passwd        | 0.0        \n",
      " mqtt.conflag.qos           | 0.0        \n",
      " mqtt.conflag.reserved      | 0.0        \n",
      " mqtt.conflag.retain        | 0.0        \n",
      " mqtt.conflag.uname         | 0.0        \n",
      " mqtt.conflag.willflag      | 0.0        \n",
      " mqtt.conflags              | 0          \n",
      " mqtt.dupflag               | 0.0        \n",
      " mqtt.hdrflags              | 0x00000030 \n",
      " mqtt.kalive                | 0.0        \n",
      " mqtt.len                   | 8.0        \n",
      " mqtt.msg                   | 32         \n",
      " mqtt.msgid                 | 0.0        \n",
      " mqtt.msgtype               | 3.0        \n",
      " mqtt.proto_len             | 0.0        \n",
      " mqtt.protoname             | 0          \n",
      " mqtt.qos                   | 0.0        \n",
      " mqtt.retain                | 0.0        \n",
      " mqtt.sub.qos               | 0.0        \n",
      " mqtt.suback.qos            | 0.0        \n",
      " mqtt.ver                   | 0.0        \n",
      " mqtt.willmsg               | 0.0        \n",
      " mqtt.willmsg_len           | 0.0        \n",
      " mqtt.willtopic             | 0.0        \n",
      " mqtt.willtopic_len         | 0.0        \n",
      " target                     | legitimate \n",
      " is_train                   | 1          \n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_Train.show(1, vertical=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Rename columns \n",
    "for column_name in df_Train.columns:\n",
    "    new_column_name = column_name.replace(\".\", \"_\")\n",
    "    df_Train = df_Train.withColumnRenamed(column_name, new_column_name)\n",
    "\n",
    "for column_name in df_Test.columns:\n",
    "    new_column_name = column_name.replace(\".\", \"_\")\n",
    "    df_Test = df_Test.withColumnRenamed(column_name, new_column_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-RECORD 0--------------------------------\n",
      " tcp_flags                  | 0x00000018 \n",
      " tcp_time_delta             | 0.998867   \n",
      " tcp_len                    | 10         \n",
      " mqtt_conack_flags          | 0          \n",
      " mqtt_conack_flags_reserved | 0.0        \n",
      " mqtt_conack_flags_sp       | 0.0        \n",
      " mqtt_conack_val            | 0.0        \n",
      " mqtt_conflag_cleansess     | 0.0        \n",
      " mqtt_conflag_passwd        | 0.0        \n",
      " mqtt_conflag_qos           | 0.0        \n",
      " mqtt_conflag_reserved      | 0.0        \n",
      " mqtt_conflag_retain        | 0.0        \n",
      " mqtt_conflag_uname         | 0.0        \n",
      " mqtt_conflag_willflag      | 0.0        \n",
      " mqtt_conflags              | 0          \n",
      " mqtt_dupflag               | 0.0        \n",
      " mqtt_hdrflags              | 0x00000030 \n",
      " mqtt_kalive                | 0.0        \n",
      " mqtt_len                   | 8.0        \n",
      " mqtt_msg                   | 32         \n",
      " mqtt_msgid                 | 0.0        \n",
      " mqtt_msgtype               | 3.0        \n",
      " mqtt_proto_len             | 0.0        \n",
      " mqtt_protoname             | 0          \n",
      " mqtt_qos                   | 0.0        \n",
      " mqtt_retain                | 0.0        \n",
      " mqtt_sub_qos               | 0.0        \n",
      " mqtt_suback_qos            | 0.0        \n",
      " mqtt_ver                   | 0.0        \n",
      " mqtt_willmsg               | 0.0        \n",
      " mqtt_willmsg_len           | 0.0        \n",
      " mqtt_willtopic             | 0.0        \n",
      " mqtt_willtopic_len         | 0.0        \n",
      " target                     | legitimate \n",
      " is_train                   | 1          \n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_Train.show(1, vertical=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-RECORD 0--------------------------------\n",
      " tcp_flags                  | 0x00000014 \n",
      " tcp_time_delta             | 0.029854   \n",
      " tcp_len                    | 0          \n",
      " mqtt_conack_flags          | 0          \n",
      " mqtt_conack_flags_reserved | 0.0        \n",
      " mqtt_conack_flags_sp       | 0.0        \n",
      " mqtt_conack_val            | 0.0        \n",
      " mqtt_conflag_cleansess     | 0.0        \n",
      " mqtt_conflag_passwd        | 0.0        \n",
      " mqtt_conflag_qos           | 0.0        \n",
      " mqtt_conflag_reserved      | 0.0        \n",
      " mqtt_conflag_retain        | 0.0        \n",
      " mqtt_conflag_uname         | 0.0        \n",
      " mqtt_conflag_willflag      | 0.0        \n",
      " mqtt_conflags              | 0          \n",
      " mqtt_dupflag               | 0.0        \n",
      " mqtt_hdrflags              | 0          \n",
      " mqtt_kalive                | 0.0        \n",
      " mqtt_len                   | 0.0        \n",
      " mqtt_msg                   | 0          \n",
      " mqtt_msgid                 | 0.0        \n",
      " mqtt_msgtype               | 0.0        \n",
      " mqtt_proto_len             | 0.0        \n",
      " mqtt_protoname             | 0          \n",
      " mqtt_qos                   | 0.0        \n",
      " mqtt_retain                | 0.0        \n",
      " mqtt_sub_qos               | 0.0        \n",
      " mqtt_suback_qos            | 0.0        \n",
      " mqtt_ver                   | 0.0        \n",
      " mqtt_willmsg               | 0.0        \n",
      " mqtt_willmsg_len           | 0.0        \n",
      " mqtt_willtopic             | 0.0        \n",
      " mqtt_willtopic_len         | 0.0        \n",
      " target                     | malformed  \n",
      " is_train                   | 0          \n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_Test.show(1, vertical=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- tcp_flags: string (nullable = true)\n",
      " |-- tcp_time_delta: double (nullable = true)\n",
      " |-- tcp_len: integer (nullable = true)\n",
      " |-- mqtt_conack_flags: string (nullable = true)\n",
      " |-- mqtt_conack_flags_reserved: double (nullable = true)\n",
      " |-- mqtt_conack_flags_sp: double (nullable = true)\n",
      " |-- mqtt_conack_val: double (nullable = true)\n",
      " |-- mqtt_conflag_cleansess: double (nullable = true)\n",
      " |-- mqtt_conflag_passwd: double (nullable = true)\n",
      " |-- mqtt_conflag_qos: double (nullable = true)\n",
      " |-- mqtt_conflag_reserved: double (nullable = true)\n",
      " |-- mqtt_conflag_retain: double (nullable = true)\n",
      " |-- mqtt_conflag_uname: double (nullable = true)\n",
      " |-- mqtt_conflag_willflag: double (nullable = true)\n",
      " |-- mqtt_conflags: string (nullable = true)\n",
      " |-- mqtt_dupflag: double (nullable = true)\n",
      " |-- mqtt_hdrflags: string (nullable = true)\n",
      " |-- mqtt_kalive: double (nullable = true)\n",
      " |-- mqtt_len: double (nullable = true)\n",
      " |-- mqtt_msg: string (nullable = true)\n",
      " |-- mqtt_msgid: double (nullable = true)\n",
      " |-- mqtt_msgtype: double (nullable = true)\n",
      " |-- mqtt_proto_len: double (nullable = true)\n",
      " |-- mqtt_protoname: string (nullable = true)\n",
      " |-- mqtt_qos: double (nullable = true)\n",
      " |-- mqtt_retain: double (nullable = true)\n",
      " |-- mqtt_sub_qos: double (nullable = true)\n",
      " |-- mqtt_suback_qos: double (nullable = true)\n",
      " |-- mqtt_ver: double (nullable = true)\n",
      " |-- mqtt_willmsg: double (nullable = true)\n",
      " |-- mqtt_willmsg_len: double (nullable = true)\n",
      " |-- mqtt_willtopic: double (nullable = true)\n",
      " |-- mqtt_willtopic_len: double (nullable = true)\n",
      " |-- target: string (nullable = true)\n",
      " |-- is_train: integer (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_Train.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-RECORD 0--------------------------------\n",
      " tcp_flags                  | 0x00000018 \n",
      " tcp_time_delta             | 0.998867   \n",
      " tcp_len                    | 10         \n",
      " mqtt_conack_flags          | 0          \n",
      " mqtt_conack_flags_reserved | 0.0        \n",
      " mqtt_conack_flags_sp       | 0.0        \n",
      " mqtt_conack_val            | 0.0        \n",
      " mqtt_conflag_cleansess     | 0.0        \n",
      " mqtt_conflag_passwd        | 0.0        \n",
      " mqtt_conflag_qos           | 0.0        \n",
      " mqtt_conflag_reserved      | 0.0        \n",
      " mqtt_conflag_retain        | 0.0        \n",
      " mqtt_conflag_uname         | 0.0        \n",
      " mqtt_conflag_willflag      | 0.0        \n",
      " mqtt_conflags              | 0          \n",
      " mqtt_dupflag               | 0.0        \n",
      " mqtt_hdrflags              | 0x00000030 \n",
      " mqtt_kalive                | 0.0        \n",
      " mqtt_len                   | 8.0        \n",
      " mqtt_msg                   | 32         \n",
      " mqtt_msgid                 | 0.0        \n",
      " mqtt_msgtype               | 3.0        \n",
      " mqtt_proto_len             | 0.0        \n",
      " mqtt_protoname             | 0          \n",
      " mqtt_qos                   | 0.0        \n",
      " mqtt_retain                | 0.0        \n",
      " mqtt_sub_qos               | 0.0        \n",
      " mqtt_suback_qos            | 0.0        \n",
      " mqtt_ver                   | 0.0        \n",
      " mqtt_willmsg               | 0.0        \n",
      " mqtt_willmsg_len           | 0.0        \n",
      " mqtt_willtopic             | 0.0        \n",
      " mqtt_willtopic_len         | 0.0        \n",
      " target                     | legitimate \n",
      " is_train                   | 1          \n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "Fake_Kaggle = df_Train.union(df_Test)\n",
    "Fake_Kaggle.show(1, vertical=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------+\n",
      "|is_train| count|\n",
      "+--------+------+\n",
      "|       1|231646|\n",
      "|       0| 99290|\n",
      "+--------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Making sure the databases are combined correctly \n",
    "dataset_count = Fake_Kaggle.groupBy('is_train').count()\n",
    "dataset_count.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Upload KDD Train\n",
    "db_properties={}\n",
    "#update your db username\n",
    "db_properties['username']=\"postgres\"\n",
    "#update your db password\n",
    "db_properties['password']=\"pratik120\"\n",
    "#make sure you got the right port number here\n",
    "db_properties['url']= \"jdbc:postgresql://localhost:5432/postgres\"\n",
    "#make sure you had the Postgres JAR file in the right location\n",
    "db_properties['driver']=\"org.postgresql.Driver\"\n",
    "db_properties['table']= \"Fake_MQTT\"\n",
    "\n",
    "\n",
    "\n",
    "Fake_Kaggle.write.format(\"jdbc\")\\\n",
    ".mode(\"overwrite\")\\\n",
    ".option(\"url\", db_properties['url'])\\\n",
    ".option(\"dbtable\", db_properties['table'])\\\n",
    ".option(\"user\", db_properties['username'])\\\n",
    ".option(\"password\", db_properties['password'])\\\n",
    ".option(\"Driver\", db_properties['driver'])\\\n",
    ".save()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-RECORD 0--------------------------------\n",
      " tcp_flags                  | 0x00000018 \n",
      " tcp_time_delta             | 0.998867   \n",
      " tcp_len                    | 10         \n",
      " mqtt_conack_flags          | 0          \n",
      " mqtt_conack_flags_reserved | 0.0        \n",
      " mqtt_conack_flags_sp       | 0.0        \n",
      " mqtt_conack_val            | 0.0        \n",
      " mqtt_conflag_cleansess     | 0.0        \n",
      " mqtt_conflag_passwd        | 0.0        \n",
      " mqtt_conflag_qos           | 0.0        \n",
      " mqtt_conflag_reserved      | 0.0        \n",
      " mqtt_conflag_retain        | 0.0        \n",
      " mqtt_conflag_uname         | 0.0        \n",
      " mqtt_conflag_willflag      | 0.0        \n",
      " mqtt_conflags              | 0          \n",
      " mqtt_dupflag               | 0.0        \n",
      " mqtt_hdrflags              | 0x00000030 \n",
      " mqtt_kalive                | 0.0        \n",
      " mqtt_len                   | 8.0        \n",
      " mqtt_msg                   | 32         \n",
      " mqtt_msgid                 | 0.0        \n",
      " mqtt_msgtype               | 3.0        \n",
      " mqtt_proto_len             | 0.0        \n",
      " mqtt_protoname             | 0          \n",
      " mqtt_qos                   | 0.0        \n",
      " mqtt_retain                | 0.0        \n",
      " mqtt_sub_qos               | 0.0        \n",
      " mqtt_suback_qos            | 0.0        \n",
      " mqtt_ver                   | 0.0        \n",
      " mqtt_willmsg               | 0.0        \n",
      " mqtt_willmsg_len           | 0.0        \n",
      " mqtt_willtopic             | 0.0        \n",
      " mqtt_willtopic_len         | 0.0        \n",
      " target                     | legitimate \n",
      " is_train                   | 1          \n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "## Reading the data back\n",
    "df_read = sqlContext.read.format(\"jdbc\")\\\n",
    "    .option(\"url\", db_properties['url'])\\\n",
    "    .option(\"dbtable\", db_properties['table'])\\\n",
    "    .option(\"user\", db_properties['username'])\\\n",
    "    .option(\"password\", db_properties['password'])\\\n",
    "    .option(\"Driver\", db_properties['driver'])\\\n",
    "    .load()\n",
    "\n",
    "df_read.show(1, vertical=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Not needed for Pratik\n",
    "# df_read = Fake_Kaggle"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import col\n",
    "from pyspark.sql.types import DoubleType\n",
    "\n",
    "casted_types_df = df_read.withColumn(\"casted_mqtt_msg\", col(\"mqtt_msg\").cast(DoubleType()))\n",
    "casted_types_df = casted_types_df.drop(\"mqtt_msg\")\n",
    "casted_types_df = casted_types_df.withColumnRenamed(\"casted_mqtt_msg\", \"mqtt_msg\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+\n",
      "|mqtt_msg             |\n",
      "+---------------------+\n",
      "|32.0                 |\n",
      "|6.361653943666145E199|\n",
      "+---------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "casted_types_df.select('mqtt_msg').show(2, truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-RECORD 0-------------------------\n",
      " tcp_flags                  | 0   \n",
      " tcp_time_delta             | 0   \n",
      " tcp_len                    | 0   \n",
      " mqtt_conack_flags          | 0   \n",
      " mqtt_conack_flags_reserved | 0   \n",
      " mqtt_conack_flags_sp       | 0   \n",
      " mqtt_conack_val            | 0   \n",
      " mqtt_conflag_cleansess     | 0   \n",
      " mqtt_conflag_passwd        | 0   \n",
      " mqtt_conflag_qos           | 0   \n",
      " mqtt_conflag_reserved      | 0   \n",
      " mqtt_conflag_retain        | 0   \n",
      " mqtt_conflag_uname         | 0   \n",
      " mqtt_conflag_willflag      | 0   \n",
      " mqtt_conflags              | 0   \n",
      " mqtt_dupflag               | 0   \n",
      " mqtt_hdrflags              | 0   \n",
      " mqtt_kalive                | 0   \n",
      " mqtt_len                   | 0   \n",
      " mqtt_msgid                 | 0   \n",
      " mqtt_msgtype               | 0   \n",
      " mqtt_proto_len             | 0   \n",
      " mqtt_protoname             | 0   \n",
      " mqtt_qos                   | 0   \n",
      " mqtt_retain                | 0   \n",
      " mqtt_sub_qos               | 0   \n",
      " mqtt_suback_qos            | 0   \n",
      " mqtt_ver                   | 0   \n",
      " mqtt_willmsg               | 0   \n",
      " mqtt_willmsg_len           | 0   \n",
      " mqtt_willtopic             | 0   \n",
      " mqtt_willtopic_len         | 0   \n",
      " target                     | 0   \n",
      " is_train                   | 0   \n",
      " mqtt_msg                   | 190 \n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Checking for Null and Nan values\n",
    "from pyspark.sql.functions import *\n",
    "null_counts_plays_df = casted_types_df.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) \\\n",
    "                                               for c in casted_types_df.columns])\n",
    "\n",
    "null_counts_plays_df.show(truncate=False, vertical=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-RECORD 0-------------------------\n",
      " tcp_flags                  | 0   \n",
      " tcp_time_delta             | 0   \n",
      " tcp_len                    | 0   \n",
      " mqtt_conack_flags          | 0   \n",
      " mqtt_conack_flags_reserved | 0   \n",
      " mqtt_conack_flags_sp       | 0   \n",
      " mqtt_conack_val            | 0   \n",
      " mqtt_conflag_cleansess     | 0   \n",
      " mqtt_conflag_passwd        | 0   \n",
      " mqtt_conflag_qos           | 0   \n",
      " mqtt_conflag_reserved      | 0   \n",
      " mqtt_conflag_retain        | 0   \n",
      " mqtt_conflag_uname         | 0   \n",
      " mqtt_conflag_willflag      | 0   \n",
      " mqtt_conflags              | 0   \n",
      " mqtt_dupflag               | 0   \n",
      " mqtt_hdrflags              | 0   \n",
      " mqtt_kalive                | 0   \n",
      " mqtt_len                   | 0   \n",
      " mqtt_msgid                 | 0   \n",
      " mqtt_msgtype               | 0   \n",
      " mqtt_proto_len             | 0   \n",
      " mqtt_protoname             | 0   \n",
      " mqtt_qos                   | 0   \n",
      " mqtt_retain                | 0   \n",
      " mqtt_sub_qos               | 0   \n",
      " mqtt_suback_qos            | 0   \n",
      " mqtt_ver                   | 0   \n",
      " mqtt_willmsg               | 0   \n",
      " mqtt_willmsg_len           | 0   \n",
      " mqtt_willtopic             | 0   \n",
      " mqtt_willtopic_len         | 0   \n",
      " target                     | 0   \n",
      " is_train                   | 0   \n",
      " mqtt_msg                   | 0   \n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Removing Rows with Null value\n",
    "casted_types_df_with_na_dropped_rows = casted_types_df.na.drop()\n",
    "\n",
    "#Check for Nan and NUll values again\n",
    "null_counts_plays_df = casted_types_df_with_na_dropped_rows.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) \\\n",
    "                        for c in casted_types_df_with_na_dropped_rows.columns])\n",
    "\n",
    "null_counts_plays_df.show(truncate=False, vertical=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-RECORD 0-------------------------\n",
      " tcp_flags                  | 0   \n",
      " tcp_time_delta             | 0   \n",
      " tcp_len                    | 0   \n",
      " mqtt_conack_flags          | 0   \n",
      " mqtt_conack_flags_reserved | 0   \n",
      " mqtt_conack_flags_sp       | 0   \n",
      " mqtt_conack_val            | 0   \n",
      " mqtt_conflag_cleansess     | 0   \n",
      " mqtt_conflag_passwd        | 0   \n",
      " mqtt_conflag_qos           | 0   \n",
      " mqtt_conflag_reserved      | 0   \n",
      " mqtt_conflag_retain        | 0   \n",
      " mqtt_conflag_uname         | 0   \n",
      " mqtt_conflag_willflag      | 0   \n",
      " mqtt_conflags              | 0   \n",
      " mqtt_dupflag               | 0   \n",
      " mqtt_hdrflags              | 0   \n",
      " mqtt_kalive                | 0   \n",
      " mqtt_len                   | 0   \n",
      " mqtt_msgid                 | 0   \n",
      " mqtt_msgtype               | 0   \n",
      " mqtt_proto_len             | 0   \n",
      " mqtt_protoname             | 0   \n",
      " mqtt_qos                   | 0   \n",
      " mqtt_retain                | 0   \n",
      " mqtt_sub_qos               | 0   \n",
      " mqtt_suback_qos            | 0   \n",
      " mqtt_ver                   | 0   \n",
      " mqtt_willmsg               | 0   \n",
      " mqtt_willmsg_len           | 0   \n",
      " mqtt_willtopic             | 0   \n",
      " mqtt_willtopic_len         | 0   \n",
      " target                     | 0   \n",
      " is_train                   | 0   \n",
      " mqtt_msg                   | 0   \n",
      "\n"
     ]
    }
   ],
   "source": [
    "casted_types_df_with_na_inf_dropped_rows=casted_types_df.filter(~(casted_types_df.mqtt_msg==float('inf')))\n",
    "null_counts_plays_df = casted_types_df_with_na_inf_dropped_rows.select([count(when(col(c)==float('inf'), c)).alias(c) for c in casted_types_df_with_na_inf_dropped_rows.columns])\n",
    "null_counts_plays_df.show(truncate=False, vertical=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2> Task 2 Part 1</h2>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Mean/ Average Length of an MQTT message: 6.55164333745422e+234\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import mean as _mean, avg as _avg, stddev as _stddev, col\n",
    "\n",
    "df_stats = casted_types_df_with_na_inf_dropped_rows.select(_avg(casted_types_df_with_na_inf_dropped_rows.mqtt_msg).alias('mean')).collect()\n",
    "print(\"Mean/ Average Length of an MQTT message:\" , df_stats[0]['mean'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['slowite', 'bruteforce', 'flood', 'malformed', 'dos', 'legitimate']\n"
     ]
    }
   ],
   "source": [
    "target_types = casted_types_df_with_na_inf_dropped_rows.select(\"target\").distinct().rdd.map(lambda r: r[0]).collect()\n",
    "print(target_types)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2>Task 2 Part 2</h2>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "avg_lens = []\n",
    "for type in target_types:\n",
    "    df_filtered = casted_types_df_with_na_inf_dropped_rows.filter(casted_types_df_with_na_inf_dropped_rows.target==type)\n",
    "    df_stats = df_filtered.select(_avg(df_filtered.tcp_len).alias('mean')).collect()\n",
    "    avg_lens.append(df_stats[0]['mean'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Average TCP length for    slowite:      4.00\n",
      "Average TCP length for bruteforce:      3.37\n",
      "Average TCP length for      flood:  11108.76\n",
      "Average TCP length for  malformed:     13.33\n",
      "Average TCP length for        dos:    312.66\n",
      "Average TCP length for legitimate:      7.77\n"
     ]
    }
   ],
   "source": [
    "for i in range(len(target_types)): \n",
    "    print(\"Average TCP length for \" + f\"{target_types[i]}\".rjust(10) + \":\" + f\"{(avg_lens[i]):.2f}\".rjust(10))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2>Task 2 Part 3</h2>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import dense_rank, rank\n",
    "from pyspark.sql import Window, types\n",
    "\n",
    "def getNMostFrequent(N, df=casted_types_df, col_name='tcp_flags'):\n",
    "    # Find unique TCP Flags\n",
    "    tcp_flags = df.select(col_name).distinct().rdd.map(lambda r: r[0]).collect()\n",
    "\n",
    "    # Count how many of each flag are present in the dataset\n",
    "    num_flags = []\n",
    "    for flag in tcp_flags:\n",
    "        num_flags.append(df.filter(df[col_name]==flag).count())\n",
    "\n",
    "    # Get the indicies of the flag count in ascending order\n",
    "    num_flags_sort_inds = [i for i, x in sorted(enumerate(num_flags), key=lambda x: x[1])]\n",
    "    num_flags_sort_inds = num_flags_sort_inds[::-1]\n",
    "\n",
    "    # Make a new array of the sorted flags\n",
    "    sorted_flags = tcp_flags.copy()\n",
    "    for i in range(len(num_flags_sort_inds)):\n",
    "        sorted_flags[i] = tcp_flags[num_flags_sort_inds[i]]\n",
    "\n",
    "    # If more flags than are present are requested, return all the sorted flags\n",
    "    if N > len(tcp_flags):\n",
    "        return sorted_flags\n",
    "\n",
    "    # Grab the N highest flags\n",
    "    n_highest_flags = sorted_flags[:N]\n",
    "\n",
    "    # Look for ties in the Nth place to return those as well\n",
    "    for i in range(N, len(tcp_flags)):\n",
    "        if not num_flags[num_flags_sort_inds[i]] == num_flags[num_flags_sort_inds[N-1]]:\n",
    "            break;\n",
    "        n_highest_flags.append(sorted_flags[i])\n",
    "\n",
    "    return n_highest_flags"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['0x00000018', '0x00000010']"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "getNMostFrequent(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2>Task 2 Part 4</h2>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+\n",
      "|key|value|\n",
      "+---+-----+\n",
      "+---+-----+\n",
      "\n",
      "Waiting...\n",
      "Waiting...\n",
      "Waiting...\n",
      "Waiting...\n",
      "Waiting...\n",
      "Waiting...\n",
      "Waiting...\n",
      "Waiting...\n",
      "Waiting...\n",
      "Waiting...\n",
      "Waiting...\n",
      "Waiting...\n",
      "Waiting...\n",
      "Waiting...\n",
      "Waiting...\n",
      "+--------------------+--------------------+\n",
      "|                 key|               value|\n",
      "+--------------------+--------------------+\n",
      "|ceos lack confide...|https://news.goog...|\n",
      "|attorney general ...|https://news.goog...|\n",
      "|national retails ...|https://news.goog...|\n",
      "|sergey kondratenk...|https://news.goog...|\n",
      "|cisos watch out t...|https://news.goog...|\n",
      "|blackpoint cyber ...|https://news.goog...|\n",
      "|survey 97 face ch...|https://news.goog...|\n",
      "|canadian organiza...|https://news.goog...|\n",
      "|what is business ...|https://news.goog...|\n",
      "|chatgpt fraudgpt ...|https://news.goog...|\n",
      "|http2 rapid reset...|https://news.goog...|\n",
      "|protecting your s...|https://news.goog...|\n",
      "|build a cyberatta...|https://news.goog...|\n",
      "|cyber security br...|https://news.goog...|\n",
      "|the biggest cyber...|https://news.goog...|\n",
      "|cybersecurity tre...|https://news.goog...|\n",
      "|todays cyber thre...|https://news.goog...|\n",
      "|the devastating b...|https://news.goog...|\n",
      "|israel is facing ...|https://news.goog...|\n",
      "|half of cybersecu...|https://news.goog...|\n",
      "+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Google news Target \n",
    "\n",
    "# Kafka Consumer\n",
    "\n",
    "from confluent_kafka import Consumer\n",
    "import socket\n",
    "from pyspark.sql.types import *\n",
    "import string\n",
    "\n",
    "\n",
    "KAFKA_CONFIG = {\n",
    "    \"bootstrap.servers\":\"pkc-6ojv2.us-west4.gcp.confluent.cloud:9092\",\n",
    "    \"security.protocol\":\"SASL_SSL\",\n",
    "    \"sasl.mechanisms\":\"PLAIN\",\n",
    "    \"sasl.username\":\"ZIGFIPPZDQBDNKRN\",\n",
    "    \"sasl.password\":\"f8gyTRboB8kvt6OXO8GmjGI4sUdbC72C2avIJCr9FYsZYBmKagc+ljNQoQJnLs9m\",\n",
    "    \"session.timeout.ms\":\"45000\",\n",
    "    \"group.id\":\"python-group-1\",\n",
    "    'auto.offset.reset': 'smallest',\n",
    "    'client.id': socket.gethostname()\n",
    "}\n",
    "\n",
    "topic_name = \"pyspark_topic\"\n",
    "\n",
    "# Clean the punctation by making a translation table that maps punctations to empty strings\n",
    "translator = str.maketrans(\"\", \"\", string.punctuation)\n",
    "\n",
    "\n",
    "emp_RDD = spark.sparkContext.emptyRDD()\n",
    "# Defining the schema of the DataFrame\n",
    "columns = StructType([StructField('key', StringType(), False),\n",
    "                      StructField('value', StringType(), False)])\n",
    "\n",
    "# Creating an empty DataFrame\n",
    "df = spark.createDataFrame(data=emp_RDD,\n",
    "                                   schema=columns)\n",
    " \n",
    "# Printing the DataFrame with no data\n",
    "df.show()\n",
    "\n",
    "consumer = Consumer(KAFKA_CONFIG)\n",
    "consumer.subscribe([topic_name])\n",
    "\n",
    "try:\n",
    "    i = 0\n",
    "    while i < 15:\n",
    "        msg = consumer.poll(timeout=1.0)\n",
    "        if msg is None:\n",
    "            i = i + 1\n",
    "            print(\"Waiting...\")\n",
    "            continue\n",
    "        if msg is not None:\n",
    "            key = msg.key().decode('utf-8').lower().translate(translator)\n",
    "            cleaned_key = \" \".join(key.split())\n",
    "            value = msg.value().decode('utf-8')\n",
    "            added_row = [[cleaned_key,value]]\n",
    "            added_df = spark.createDataFrame(added_row, columns)\n",
    "            df = df.union(added_df)\n",
    "\n",
    "except KeyboardInterrupt:\n",
    "    pass\n",
    "finally:\n",
    "    consumer.close()\n",
    "    df.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+\n",
      "|word|count|\n",
      "+----+-----+\n",
      "+----+-----+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "----------------------------------------\n",
      "Exception occurred during processing of request from ('127.0.0.1', 61611)\n",
      "Traceback (most recent call last):\n",
      "  File \"c:\\Users\\prati\\anaconda3\\lib\\socketserver.py\", line 316, in _handle_request_noblock\n",
      "    self.process_request(request, client_address)\n",
      "  File \"c:\\Users\\prati\\anaconda3\\lib\\socketserver.py\", line 347, in process_request\n",
      "    self.finish_request(request, client_address)\n",
      "  File \"c:\\Users\\prati\\anaconda3\\lib\\socketserver.py\", line 360, in finish_request\n",
      "    self.RequestHandlerClass(request, client_address, self)\n",
      "  File \"c:\\Users\\prati\\anaconda3\\lib\\socketserver.py\", line 747, in __init__\n",
      "    self.handle()\n",
      "  File \"C:\\spark\\python\\pyspark\\accumulators.py\", line 295, in handle\n",
      "    poll(accum_updates)\n",
      "  File \"C:\\spark\\python\\pyspark\\accumulators.py\", line 267, in poll\n",
      "    if self.rfile in r and func():\n",
      "  File \"C:\\spark\\python\\pyspark\\accumulators.py\", line 271, in accum_updates\n",
      "    num_updates = read_int(self.rfile)\n",
      "  File \"C:\\spark\\python\\pyspark\\serializers.py\", line 594, in read_int\n",
      "    length = stream.read(4)\n",
      "  File \"c:\\Users\\prati\\anaconda3\\lib\\socket.py\", line 704, in readinto\n",
      "    return self._sock.recv_into(b)\n",
      "ConnectionResetError: [WinError 10054] An existing connection was forcibly closed by the remote host\n",
      "----------------------------------------\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import *\n",
    "\n",
    "streamed_data = df.withColumn('word', explode(split(col('key'), ' '))) \\\n",
    "                .filter(col('word').isin(target_types)) \\\n",
    "                .groupBy('word') \\\n",
    "                .count() \\\n",
    "                .sort('count', ascending=False)\n",
    "    \n",
    "streamed_data.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.9.17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}