Project.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "508e2ee9",
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    },
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Ivy Default Cache set to: /root/.ivy2/cache\n",
      "The jars for the packages stored in: /root/.ivy2/jars\n",
      ":: loading settings :: url = jar:file:/usr/local/lib/python3.9/dist-packages/pyspark/jars/ivy-2.4.0.jar!/org/apache/ivy/core/settings/ivysettings.xml\n",
      "org.apache.spark#spark-sql-kafka-0-10_2.12 added as a dependency\n",
      ":: resolving dependencies :: org.apache.spark#spark-submit-parent-4393facb-ccdd-479f-9b9c-0a00d1d6e1cf;1.0\n",
      "\tconfs: [default]\n",
      "\tfound org.apache.spark#spark-sql-kafka-0-10_2.12;3.0.0 in central\n",
      "\tfound org.apache.spark#spark-token-provider-kafka-0-10_2.12;3.0.0 in central\n",
      "\tfound org.apache.kafka#kafka-clients;2.4.1 in central\n",
      "\tfound com.github.luben#zstd-jni;1.4.4-3 in central\n",
      "\tfound org.lz4#lz4-java;1.7.1 in central\n",
      "\tfound org.xerial.snappy#snappy-java;1.1.7.5 in central\n",
      "\tfound org.slf4j#slf4j-api;1.7.30 in central\n",
      "\tfound org.spark-project.spark#unused;1.0.0 in central\n",
      "\tfound org.apache.commons#commons-pool2;2.6.2 in central\n",
      ":: resolution report :: resolve 631ms :: artifacts dl 25ms\n",
      "\t:: modules in use:\n",
      "\tcom.github.luben#zstd-jni;1.4.4-3 from central in [default]\n",
      "\torg.apache.commons#commons-pool2;2.6.2 from central in [default]\n",
      "\torg.apache.kafka#kafka-clients;2.4.1 from central in [default]\n",
      "\torg.apache.spark#spark-sql-kafka-0-10_2.12;3.0.0 from central in [default]\n",
      "\torg.apache.spark#spark-token-provider-kafka-0-10_2.12;3.0.0 from central in [default]\n",
      "\torg.lz4#lz4-java;1.7.1 from central in [default]\n",
      "\torg.slf4j#slf4j-api;1.7.30 from central in [default]\n",
      "\torg.spark-project.spark#unused;1.0.0 from central in [default]\n",
      "\torg.xerial.snappy#snappy-java;1.1.7.5 from central in [default]\n",
      "\t---------------------------------------------------------------------\n",
      "\t|                  |            modules            ||   artifacts   |\n",
      "\t|       conf       | number| search|dwnlded|evicted|| number|dwnlded|\n",
      "\t---------------------------------------------------------------------\n",
      "\t|      default     |   9   |   0   |   0   |   0   ||   9   |   0   |\n",
      "\t---------------------------------------------------------------------\n",
      ":: retrieving :: org.apache.spark#spark-submit-parent-4393facb-ccdd-479f-9b9c-0a00d1d6e1cf\n",
      "\tconfs: [default]\n",
      "\t0 artifacts copied, 9 already retrieved (0kB/12ms)\n",
      "23/05/01 16:54:05 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.\\\n",
    "        builder.\\\n",
    "        appName(\"pyspark-kafka-streaming\").\\\n",
    "        master(\"spark://spark-master:7077\").\\\n",
    "        config(\"spark.jars.packages\", \"org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.0\"). \\\n",
    "        config(\"spark.executor.memory\", \"512m\").\\\n",
    "        getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "619e558e",
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "df_streamed_raw = (spark\n",
    "  .readStream\n",
    "  # Add your code here\n",
    "  .option(\"kafka.bootstrap.servers\", \"kafka:9093\")\n",
    "  .format(\"kafka\")\n",
    "  .option(\"subscribe\", \"topic\")\n",
    "  .load())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "2a2a38c0",
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    },
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/05/01 16:54:24 WARN StreamingQueryManager: Temporary checkpoint location created which is deleted normally when the query didn't fail: /tmp/temporary-dd0cd504-6e31-4cc6-b0dc-b8da74a02caf. If it's required to delete it under any circumstances, please set spark.sql.streaming.forceDeleteTempCheckpointLocation to true. Important to know deleting temp checkpoint folder is best effort.\n",
      "[Stage 0:>                                                          (0 + 1) / 1]\r"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.types import StringType, ArrayType, FloatType, IntegerType\n",
    "from pyspark.sql.functions import *\n",
    "\n",
    "df_streamed_kv = (df_streamed_raw\n",
    "    .withColumn(\"key\", df_streamed_raw[\"key\"].cast(StringType()))\n",
    "    .withColumn(\"value\", df_streamed_raw[\"value\"].cast(StringType())))\n",
    "\n",
    "test_query = (df_streamed_kv\n",
    "                .writeStream\n",
    "                .format(\"memory\")\n",
    "              .outputMode(\"update\")\n",
    "              .queryName(\"test_query_table\")\n",
    "                .start())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "c8de6860",
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    },
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+-----+---------+------+--------------------+-------------+\n",
      "|key|               value|topic|partition|offset|           timestamp|timestampType|\n",
      "+---+--------------------+-----+---------+------+--------------------+-------------+\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24202|2023-05-01 16:54:...|            0|\n",
      "|key|                null|topic|        0| 24203|2023-05-01 16:54:...|            0|\n",
      "|key|                null|topic|        0| 24204|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24205|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24206|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24207|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24208|2023-05-01 16:54:...|            0|\n",
      "|key|                null|topic|        0| 24209|2023-05-01 16:54:...|            0|\n",
      "|key|                null|topic|        0| 24210|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24211|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24212|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24213|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24214|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24215|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24216|2023-05-01 16:54:...|            0|\n",
      "|key|                null|topic|        0| 24217|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24218|2023-05-01 16:54:...|            0|\n",
      "|key|                null|topic|        0| 24219|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24220|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24221|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24222|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24223|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24224|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24225|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24226|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24227|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24228|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24229|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24230|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24231|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24232|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24233|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24234|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24235| 2023-05-01 16:54:36|            0|\n",
      "|key|                null|topic|        0| 24236|2023-05-01 16:54:...|            0|\n",
      "|key|                null|topic|        0| 24237|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24238|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24239|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24240|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24241|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24242|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24243|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24244|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24245|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24246|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24247|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24248|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24249|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24250|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24251|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24252|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24253|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24254|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24255|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24256|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24257|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24258|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24259|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24260|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24261|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24262|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24263|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24264|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24265|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24266|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24267|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24268|2023-05-01 16:54:...|            0|\n",
      "|key|                null|topic|        0| 24269|2023-05-01 16:54:...|            0|\n",
      "|key|                null|topic|        0| 24270|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24271|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24272|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24273|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24274|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24275|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24276|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24277|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24278|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24279|2023-05-01 16:54:...|            0|\n",
      "|key|{\"timestamp\": 167...|topic|        0| 24280|2023-05-01 16:54:...|            0|\n",
      "+---+--------------------+-----+---------+------+--------------------+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"Select * from test_query_table\").show(100)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6305fee8",
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "source": [
    "test_query.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "feeac8ac",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_anti_null  = df_streamed_kv.filter(\"value IS NOT NULL\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "5ffe0afe",
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import from_json\n",
    "from pyspark.sql.types import StructType, StructField, BooleanType, LongType, IntegerType\n",
    "\n",
    "\n",
    "event_schema = StructType([\n",
    "    StructField(\"timestamp\", StringType()),\n",
    "    StructField(\"process_id\", StringType()),\n",
    "    StructField(\"username\", StringType()),\n",
    "    StructField(\"ip\", StringType()),\n",
    "    StructField(\"is_private\", StringType()),\n",
    "    StructField(\"is_root\", StringType()),\n",
    "    StructField(\"is_failure\", StringType()),\n",
    "    StructField(\"time_since_last_failure\", StringType()),\n",
    "    StructField(\"time_since_last_failure_of_same_type\", StringType()),\n",
    "    StructField(\"failure_count_in_last_15_mins\", StringType()),\n",
    "    StructField(\"failure_count_in_last_30_mins\", StringType()),\n",
    "    StructField(\"failure_count_in_last_60_mins\", StringType()),\n",
    "    StructField(\"label_auth_failure\", StringType()),\n",
    "    StructField(\"label_break_in_attempt\", StringType()),\n",
    "    StructField(\"label_connection_closed\", StringType()),\n",
    "    StructField(\"label_disconnect\", StringType()),\n",
    "    StructField(\"label_failed_password\", StringType()),\n",
    "    StructField(\"label_invalid_user\", StringType()),\n",
    "    StructField(\"label_no_label\", StringType()),\n",
    "    StructField(\"label_no_identification\", StringType()),\n",
    "    StructField(\"class\", StringType())\n",
    "])\n",
    "\n",
    "# Parse the events from JSON format\n",
    "df_parsed = (df_anti_null\n",
    "           # Sets schema for event data\n",
    "           .withColumn(\"value\", from_json(\"value\", event_schema))\n",
    "          )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "77499766",
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "df_formatted = (df_parsed.select(\n",
    "#     col(\"key\").alias(\"event_key\")\n",
    "#     ,col(\"topic\").alias(\"event_topic\")\n",
    "# #     ,col(\"timestamp\").alias(\"event_timestamp\")\n",
    "#     ,col(\"value.timestamp\")\n",
    "#     ,col(\"value.process_id\")\n",
    "#     ,col(\"value.username\")\n",
    "#     ,col(\"value.ip\")\n",
    "    col(\"value.is_private\").cast(FloatType())\n",
    "    ,col(\"value.is_root\").cast(FloatType())\n",
    "    ,col(\"value.is_failure\").cast(FloatType())\n",
    "#     ,col(\"value.time_since_last_failure\")\n",
    "    ,col(\"value.time_since_last_failure_of_same_type\").cast(FloatType())\n",
    "    ,col(\"value.failure_count_in_last_15_mins\").cast(FloatType())\n",
    "    ,col(\"value.failure_count_in_last_30_mins\").cast(FloatType())\n",
    "    ,col(\"value.failure_count_in_last_60_mins\").cast(FloatType())\n",
    "    ,col(\"value.label_auth_failure\").cast(FloatType())\n",
    "    ,col(\"value.label_break_in_attempt\").cast(FloatType())\n",
    "    ,col(\"value.label_connection_closed\").cast(FloatType())\n",
    "    ,col(\"value.label_disconnect\").cast(FloatType())\n",
    "    ,col(\"value.label_failed_password\").cast(FloatType())\n",
    "    ,col(\"value.label_invalid_user\").cast(FloatType())\n",
    "    ,col(\"value.label_no_label\").cast(FloatType())\n",
    "    ,col(\"value.label_no_identification\").cast(FloatType())\n",
    "    ,col(\"value.class\").cast(FloatType())\n",
    "))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "572f4bd1",
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[is_private: float, is_root: float, is_failure: float, time_since_last_failure_of_same_type: float, failure_count_in_last_15_mins: float, failure_count_in_last_30_mins: float, failure_count_in_last_60_mins: float, label_auth_failure: float, label_break_in_attempt: float, label_connection_closed: float, label_disconnect: float, label_failed_password: float, label_invalid_user: float, label_no_label: float, label_no_identification: float, class: float]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_formatted"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "67312ca1",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_format_null = df_formatted.filter(\"is_private IS NOT NULL\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "5f763662",
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    },
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/05/01 16:57:19 WARN StreamingQueryManager: Temporary checkpoint location created which is deleted normally when the query didn't fail: /tmp/temporary-56600c33-5487-452c-8d8c-18f020f660df. If it's required to delete it under any circumstances, please set spark.sql.streaming.forceDeleteTempCheckpointLocation to true. Important to know deleting temp checkpoint folder is best effort.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 0\n",
      "-------------------------------------------\n",
      "+----------+-------+----------+------------------------------------+-----------------------------+-----------------------------+-----------------------------+------------------+----------------------+-----------------------+----------------+---------------------+------------------+--------------+-----------------------+-----+\n",
      "|is_private|is_root|is_failure|time_since_last_failure_of_same_type|failure_count_in_last_15_mins|failure_count_in_last_30_mins|failure_count_in_last_60_mins|label_auth_failure|label_break_in_attempt|label_connection_closed|label_disconnect|label_failed_password|label_invalid_user|label_no_label|label_no_identification|class|\n",
      "+----------+-------+----------+------------------------------------+-----------------------------+-----------------------------+-----------------------------+------------------+----------------------+-----------------------+----------------+---------------------+------------------+--------------+-----------------------+-----+\n",
      "+----------+-------+----------+------------------------------------+-----------------------------+-----------------------------+-----------------------------+------------------+----------------------+-----------------------+----------------+---------------------+------------------+--------------+-----------------------+-----+\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 1\n",
      "-------------------------------------------\n",
      "+----------+-------+----------+------------------------------------+-----------------------------+-----------------------------+-----------------------------+------------------+----------------------+-----------------------+----------------+---------------------+------------------+--------------+-----------------------+-----+\n",
      "|is_private|is_root|is_failure|time_since_last_failure_of_same_type|failure_count_in_last_15_mins|failure_count_in_last_30_mins|failure_count_in_last_60_mins|label_auth_failure|label_break_in_attempt|label_connection_closed|label_disconnect|label_failed_password|label_invalid_user|label_no_label|label_no_identification|class|\n",
      "+----------+-------+----------+------------------------------------+-----------------------------+-----------------------------+-----------------------------+------------------+----------------------+-----------------------+----------------+---------------------+------------------+--------------+-----------------------+-----+\n",
      "|       0.0|    0.0|       0.0|                                 0.0|                          0.0|                          0.0|                          0.0|               0.0|                   0.0|                    0.0|             1.0|                  0.0|               0.0|           0.0|                    0.0|  0.0|\n",
      "|       0.0|    0.0|       1.0|                                 0.0|                         41.0|                         41.0|                         41.0|               0.0|                   1.0|                    0.0|             0.0|                  0.0|               0.0|           0.0|                    0.0|  1.0|\n",
      "|       0.0|    1.0|       1.0|                                 5.0|                         42.0|                         42.0|                         42.0|               1.0|                   0.0|                    0.0|             0.0|                  0.0|               0.0|           0.0|                    0.0|  0.0|\n",
      "+----------+-------+----------+------------------------------------+-----------------------------+-----------------------------+-----------------------------+------------------+----------------------+-----------------------+----------------+---------------------+------------------+--------------+-----------------------+-----+\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 2\n",
      "-------------------------------------------\n",
      "+----------+-------+----------+------------------------------------+-----------------------------+-----------------------------+-----------------------------+------------------+----------------------+-----------------------+----------------+---------------------+------------------+--------------+-----------------------+-----+\n",
      "|is_private|is_root|is_failure|time_since_last_failure_of_same_type|failure_count_in_last_15_mins|failure_count_in_last_30_mins|failure_count_in_last_60_mins|label_auth_failure|label_break_in_attempt|label_connection_closed|label_disconnect|label_failed_password|label_invalid_user|label_no_label|label_no_identification|class|\n",
      "+----------+-------+----------+------------------------------------+-----------------------------+-----------------------------+-----------------------------+------------------+----------------------+-----------------------+----------------+---------------------+------------------+--------------+-----------------------+-----+\n",
      "|       0.0|    1.0|       1.0|                                10.0|                         43.0|                         43.0|                         43.0|               0.0|                   0.0|                    0.0|             0.0|                  1.0|               0.0|           0.0|                    0.0|  0.0|\n",
      "|       0.0|    0.0|       0.0|                                 0.0|                          0.0|                          0.0|                          0.0|               0.0|                   0.0|                    0.0|             1.0|                  0.0|               0.0|           0.0|                    0.0|  0.0|\n",
      "|       0.0|    0.0|       1.0|                                 0.0|                         44.0|                         44.0|                         44.0|               0.0|                   1.0|                    0.0|             0.0|                  0.0|               0.0|           0.0|                    0.0|  1.0|\n",
      "|       0.0|    1.0|       1.0|                                 6.0|                         45.0|                         45.0|                         45.0|               1.0|                   0.0|                    0.0|             0.0|                  0.0|               0.0|           0.0|                    0.0|  0.0|\n",
      "|       0.0|    1.0|       1.0|                                 5.0|                         46.0|                         46.0|                         46.0|               0.0|                   0.0|                    0.0|             0.0|                  1.0|               0.0|           0.0|                    0.0|  0.0|\n",
      "|       0.0|    0.0|       0.0|                                 0.0|                          0.0|                          0.0|                          0.0|               0.0|                   0.0|                    0.0|             1.0|                  0.0|               0.0|           0.0|                    0.0|  0.0|\n",
      "+----------+-------+----------+------------------------------------+-----------------------------+-----------------------------+-----------------------------+------------------+----------------------+-----------------------+----------------+---------------------+------------------+--------------+-----------------------+-----+\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 3\n",
      "-------------------------------------------\n",
      "+----------+-------+----------+------------------------------------+-----------------------------+-----------------------------+-----------------------------+------------------+----------------------+-----------------------+----------------+---------------------+------------------+--------------+-----------------------+-----+\n",
      "|is_private|is_root|is_failure|time_since_last_failure_of_same_type|failure_count_in_last_15_mins|failure_count_in_last_30_mins|failure_count_in_last_60_mins|label_auth_failure|label_break_in_attempt|label_connection_closed|label_disconnect|label_failed_password|label_invalid_user|label_no_label|label_no_identification|class|\n",
      "+----------+-------+----------+------------------------------------+-----------------------------+-----------------------------+-----------------------------+------------------+----------------------+-----------------------+----------------+---------------------+------------------+--------------+-----------------------+-----+\n",
      "|       0.0|    0.0|       1.0|                                 0.0|                         47.0|                         47.0|                         47.0|               0.0|                   1.0|                    0.0|             0.0|                  0.0|               0.0|           0.0|                    0.0|  1.0|\n",
      "|       0.0|    1.0|       1.0|                                 5.0|                         48.0|                         48.0|                         48.0|               1.0|                   0.0|                    0.0|             0.0|                  0.0|               0.0|           0.0|                    0.0|  0.0|\n",
      "|       0.0|    1.0|       1.0|                                 6.0|                         49.0|                         49.0|                         49.0|               0.0|                   0.0|                    0.0|             0.0|                  1.0|               0.0|           0.0|                    0.0|  0.0|\n",
      "|       0.0|    0.0|       0.0|                                 0.0|                          0.0|                          0.0|                          0.0|               0.0|                   0.0|                    0.0|             1.0|                  0.0|               0.0|           0.0|                    0.0|  0.0|\n",
      "|       0.0|    0.0|       1.0|                                 0.0|                         49.0|                         49.0|                         49.0|               0.0|                   1.0|                    0.0|             0.0|                  0.0|               0.0|           0.0|                    0.0|  1.0|\n",
      "|       0.0|    1.0|       1.0|                                 5.0|                         50.0|                         50.0|                         50.0|               1.0|                   0.0|                    0.0|             0.0|                  0.0|               0.0|           0.0|                    0.0|  0.0|\n",
      "|       0.0|    1.0|       1.0|                                12.0|                         51.0|                         51.0|                         51.0|               0.0|                   0.0|                    0.0|             0.0|                  1.0|               0.0|           0.0|                    0.0|  0.0|\n",
      "+----------+-------+----------+------------------------------------+-----------------------------+-----------------------------+-----------------------------+------------------+----------------------+-----------------------+----------------+---------------------+------------------+--------------+-----------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "\n",
    "query = (df_format_null\n",
    "  .writeStream\n",
    "  .format(\"console\") \\\n",
    "  .trigger(processingTime = '2 seconds')\\\n",
    " .start())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "8354fa07",
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    },
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "query.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "0a76810e",
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import VectorAssembler\n",
    "\n",
    "vecCols = ['is_private', 'is_root', 'is_failure', 'time_since_last_failure_of_same_type', 'failure_count_in_last_15_mins',\n",
    "       'failure_count_in_last_30_mins', 'failure_count_in_last_60_mins','label_auth_failure', 'label_break_in_attempt',\n",
    "       'label_connection_closed', 'label_disconnect', 'label_failed_password',\n",
    "       'label_invalid_user', 'label_no_label', 'label_no_identification']\n",
    "assembler = VectorAssembler(inputCols=vecCols, outputCol=\"vectors\")\n",
    "df_assembler = assembler.transform(df_format_null)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "73cf6c1d",
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    },
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.classification import RandomForestClassificationModel\n",
    "\n",
    "model = RandomForestClassificationModel.load('/data/ml_py_model.pkl')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "9cf83cc8",
   "metadata": {},
   "outputs": [],
   "source": [
    "pred = model.transform(df_assembler)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "299ef830",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_out = pred.select(\"prediction\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ea765c6b",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "df_out_query = (df_out\n",
    "               .writeStream\n",
    "               .format(\"console\")\n",
    "               .trigger(processingTime='5 seconds')\n",
    "                .start()\n",
    "               )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "edd653ba",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_out_query.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "c6e58cd1",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_filtered = (df_out\n",
    "              .filter(\"prediction == 1.0\")\n",
    "              )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "d7f13771",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/05/01 16:59:37 WARN StreamingQueryManager: Temporary checkpoint location created which is deleted normally when the query didn't fail: /tmp/temporary-1eff53c3-8a4a-460a-9298-17b9cc902079. If it's required to delete it under any circumstances, please set spark.sql.streaming.forceDeleteTempCheckpointLocation to true. Important to know deleting temp checkpoint folder is best effort.\n",
      "23/05/01 16:59:37 WARN package: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 0\n",
      "-------------------------------------------\n",
      "+----------+\n",
      "|prediction|\n",
      "+----------+\n",
      "+----------+\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 1\n",
      "-------------------------------------------\n",
      "+----------+\n",
      "|prediction|\n",
      "+----------+\n",
      "+----------+\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 2\n",
      "-------------------------------------------\n",
      "+----------+\n",
      "|prediction|\n",
      "+----------+\n",
      "+----------+\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 3\n",
      "-------------------------------------------\n",
      "+----------+\n",
      "|prediction|\n",
      "+----------+\n",
      "+----------+\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 4\n",
      "-------------------------------------------\n",
      "+----------+\n",
      "|prediction|\n",
      "+----------+\n",
      "|       1.0|\n",
      "+----------+\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 5\n",
      "-------------------------------------------\n",
      "+----------+\n",
      "|prediction|\n",
      "+----------+\n",
      "|       1.0|\n",
      "+----------+\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 6\n",
      "-------------------------------------------\n",
      "+----------+\n",
      "|prediction|\n",
      "+----------+\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_fil_run = (df_filtered\n",
    "             .writeStream\n",
    "             .format(\"console\")\n",
    "             .trigger(processingTime='5 seconds')\n",
    "             .start()\n",
    "             )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "60889ca3",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_fil_run.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2a6f79ae",
   "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.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}