mwu-var2.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## FAIR USE POLICY\n",
    "**Please do not leave your Jupyter lab server idle for extended periods of time.** The Jupyter process, active Python kernels, and especially running Spark contexts, claim a minimum amount of cluster resources. These add up and will get starve resources of others eventually. Leaving your environment idle for a few hours (e.g., over lunch) is fine. But letting it idle overnight or for multiple days in which you are not actively using the cluster is not. You can kill the server from your SSH session, by pressing ctrl+c repeatedly, or by selecting *File->Shutdown* from the menu."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import time\n",
    "from datetime import datetime, timedelta, date\n",
    "import pprint\n",
    "\n",
    "import collections\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import subprocess\n",
    "\n",
    "import pyspark\n",
    "# Find Spark\n",
    "import findspark\n",
    "findspark.init()\n",
    "\n",
    "\n",
    "from datetime import datetime, timedelta, date\n",
    "import datetime as dt\n",
    "from datetime import timedelta as td\n",
    "import os\n",
    "import pandas as pd\n",
    "import subprocess\n",
    "import re\n",
    "import pyarrow as pa\n",
    "import copy\n",
    "from pyspark.sql.types import DateType\n",
    "from string import digits\n",
    "from dateutil import parser\n",
    "\n",
    "import nltk\n",
    "import pyspark\n",
    "from pyspark.sql import *  \n",
    "from pyspark.sql.functions import *\n",
    "import pyspark.sql.functions as F \n",
    "from pyspark.sql import functions as F, Window\n",
    "from pyspark.sql.functions import from_unixtime\n",
    "from pyspark.sql.types import StringType\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql.functions import mean as _mean\n",
    "from pyspark.sql.window import Window as W\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql.functions import col\n",
    "from pyspark.sql.types import StructType, StructField, LongType\n",
    "from scipy import stats\n",
    "from scipy.stats import mannwhitneyu\n",
    "\n",
    "\n",
    "from functools import reduce\n",
    "from pyspark.ml.feature import Tokenizer, StopWordsRemover\n",
    "\n",
    "\n",
    "import json\n",
    "import glob\n",
    "#json.loads('{\"\":\"\\\\ud800\"}')\n",
    "\n",
    "\n",
    "\n",
    "#spark.conf.set(\"spark.sql.broadcastTimeout\", 3000)\n",
    "#spark.conf.set(\"spark.sql.autoBroadcastJoinThreshold\", -1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create Spark Configuration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a SparkConf\n",
    "APP_NAME = \"apwg-median-app\"\n",
    "\n",
    "spark_conf = pyspark.SparkConf().setAppName(APP_NAME).setMaster(\"yarn\").set(\n",
    "    \"spark.submit.deployMode\", \"client\"\n",
    ").set(\"spark.sql.parquet.binaryAsString\", \"true\"\n",
    ").set(\"spark.dynamicAllocation.maxExecutors\", \"16\"\n",
    ").set(\"spark.jars.packages\", \"com.johnsnowlabs.nlp:spark-nlp-spark24_2.11:3.0.0\"\n",
    ").set(\"spark.sql.debug.maxToStringFields\", \"1000\"\n",
    ").set(\"spark.executor.memory\", \"7G\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Start SparkContext\n",
    "1. This may take a minute to complete\n",
    "2. You should not (and cannot) start two Spark contexts. If you accidentally run this cell twice or get stuck somehow, restart your Python kernel from the menu above.\n",
    "3. Please **stop your Spark context** when idling for extended periods of time (see code at bottom of notebook)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2021-05-05 21:12:02] Starting Spark context.\n"
     ]
    }
   ],
   "source": [
    "print(\"[{}] Starting Spark context.\".format(datetime.now().replace(microsecond=0)))\n",
    "\n",
    "# SparkContext\n",
    "sc = pyspark.SparkContext(conf=spark_conf)\n",
    "\n",
    "# SQLContext\n",
    "sqlc = pyspark.SQLContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Your code below"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "APWG_CLEAN_DATA_CONVERTED_BASE = \"PATH TO DATA-RECORDS\"\n",
    "\n",
    "INTEREST_DATE_START = datetime(2017, 12, 18)\n",
    "INTEREST_DATE_END   = datetime(2019, 8, 16)\n",
    "\n",
    "# Read JSON files into Spark DF\n",
    "clean_mails_df = sqlc.read.json(APWG_CLEAN_DATA_CONVERTED_BASE, multiLine=True).withColumn(\n",
    "    \"parsed_date\", F.from_unixtime(F.col(\"modified\")).cast(\"date\")\n",
    ").filter(\n",
    "    # Filter date range of interest\n",
    "    (F.col(\"parsed_date\") >= INTEREST_DATE_START.date().isoformat()) &\n",
    "    (F.col(\"parsed_date\") <= INTEREST_DATE_END.date().isoformat())\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "#filter emails in english language\n",
    "original_df = clean_mails_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "original_df = original_df.filter(original_df.language == \"english\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "h1_df = original_df   \n",
    "h1_df = h1_df.select('parsed_date', 'id')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "remover = StopWordsRemover(inputCol='body_words', outputCol='words_clean')\n",
    "h2_df = remover.transform(original_df)\n",
    "\n",
    "h2_df = h2_df.drop(\"body_words\")\n",
    "\n",
    "word_list=['unsubscribed', 'hack', 'takedown', 'password', 'transparent',\\\n",
    "           'attempt', 'redirect', 'impersonate', 'network', 'obsolete', 'illegal', 'damage', 'edit',\\\n",
    "           'unauthenticated', 'initial', 'survey', 'collect', 'victim', 'detect', 'recharge', 'test',\\\n",
    "           'attachment', 'claim', 'profitable', 'virus', 'fraudulent', 'revalidation', 'link', 'description']\n",
    "\n",
    "#array_intersect function requires two arrays as arguments, create array from the list of given values:\n",
    "list_col = F.array(*[F.lit(cl) for cl in word_list])\n",
    "h2_df = h2_df.filter(F.size(F.array_intersect(F.col(\"words_clean\"), list_col)) > 0)\n",
    "h2_df = h2_df.select('parsed_date', 'id')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "h3_df = original_df.filter(original_df.email_has_attachments == \"1\")\n",
    "h3_df = h3_df.select('parsed_date', 'id')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# announcement date of ddos attack\n",
    "\n",
    "ddos_list = ['2018-01-17', '2018-01-27', '2018-02-08', '2018-03-01', '2018-03-06', '2018-05-14', '2018-05-24', '2018-07-30',\\\n",
    "             '2019-01-16','2019-01-31', '2019-02-23', '2019-03-22', '2019-04-16', '2019-06-12', '2019-08-08']\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "#partitionBy is used to shuffle data before applying the functions\n",
    "def hypo_1(df, day):\n",
    "    \n",
    "    h1_df1 = (df.filter(f\"parsed_date < '{day}' and parsed_date > '{day}' - interval 20 days\")\n",
    "             .withColumn('rn', F.dense_rank().over(Window.orderBy(F.desc('parsed_date'))))\n",
    "             .filter('rn <= 7')\n",
    "             .drop('rn')\n",
    "             .withColumn('count_before', F.count('id').over(Window.partitionBy('parsed_date')))\n",
    "             .orderBy('parsed_date').withColumn('ddos_date', F.lit(day))\n",
    "          )\n",
    "    #h1_df1 = h1_df1.dropDuplicates(['parsed_date', 'count_before'])\n",
    "    h1_df2 = (df.filter(f\"parsed_date < '{day}' + interval 20 days and parsed_date > '{day}'\")\n",
    "             .withColumn('rn', F.dense_rank().over(Window.orderBy('parsed_date')))\n",
    "             .filter('rn <= 7')\n",
    "             .drop('rn')\n",
    "             .withColumn('count_after', F.count('id').over(Window.partitionBy('parsed_date')))\n",
    "             .orderBy('parsed_date').withColumn('ddos_date', F.lit(day))\n",
    "          )\n",
    "    \n",
    "    \n",
    "    return [h1_df1, h1_df2]\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "def hypo_2(df, day):\n",
    "    \"\"\"\n",
    "    Example usage: df_list = hypo_2(df, '2017-12-18', 15)\n",
    "    Returns a list of 2 dataframes.\n",
    "    \"\"\"\n",
    "    h2_df1 = (df.filter(f\"parsed_date < '{day}' and parsed_date > '{day}' - interval 20 days\")\n",
    "             .withColumn('rn', F.dense_rank().over(Window.orderBy(F.desc('parsed_date'))))\n",
    "             .filter('rn <= 15')\n",
    "             .drop('rn')\n",
    "             .withColumn('count_before', F.count('id').over(Window.partitionBy('parsed_date')))\n",
    "             .orderBy('parsed_date').withColumn('ddos_date', F.lit(day))\n",
    "          )\n",
    "   \n",
    "    \n",
    "    h2_df2 = (df.filter(f\"parsed_date < '{day}' + interval 20 days and parsed_date > '{day}'\")\n",
    "             .withColumn('rn', F.dense_rank().over(Window.orderBy('parsed_date')))\n",
    "             .filter('rn <= 15')\n",
    "             .drop('rn')\n",
    "             .withColumn('count_after', F.count('id').over(Window.partitionBy('parsed_date')))\n",
    "             .orderBy('parsed_date').withColumn('ddos_date', F.lit(day))\n",
    "          )\n",
    "\n",
    "    return [h2_df1, h2_df2]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "def hypo_3(df, day):\n",
    "    \"\"\"\n",
    "    Example usage: df_list = hypo_3(df, '2017-12-18', 15)\n",
    "    Returns a list of 2 dataframes.\n",
    "    \"\"\"\n",
    "    h3_df1 = (df.filter(f\"parsed_date < '{day}' and parsed_date > '{day}' - interval 20 days\")\n",
    "             .withColumn('rn', F.dense_rank().over(Window.orderBy(F.desc('parsed_date'))))\n",
    "             .filter('rn <= 15')\n",
    "             .drop('rn')\n",
    "             .withColumn('count_before', F.count('id').over(Window.partitionBy('parsed_date')))\n",
    "             .orderBy('parsed_date').withColumn('ddos_date', F.lit(day))\n",
    "          )\n",
    "\n",
    "    \n",
    "    h3_df2 = (df.filter(f\"parsed_date < '{day}' + interval 20 days and parsed_date > '{day}'\")\n",
    "             .withColumn('rn', F.dense_rank().over(Window.orderBy('parsed_date')))\n",
    "             .filter('rn <= 15')\n",
    "             .drop('rn')\n",
    "             .withColumn('count_after', F.count('id').over(Window.partitionBy('parsed_date')))\n",
    "             .orderBy('parsed_date').withColumn('ddos_date', F.lit(day))\n",
    "          )\n",
    "   \n",
    "    \n",
    "    \n",
    "    return [h3_df1, h3_df2]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "h1_df1_list = []\n",
    "h1_df2_list = []\n",
    "\n",
    "for d in ddos_list:\n",
    "    h1_df1, h1_df2 = hypo_1(h1_df, d)\n",
    "    \n",
    "    h1_df1 = h1_df1.dropDuplicates(['parsed_date', 'count_before'])\n",
    "    h1_df2 = h1_df2.dropDuplicates(['parsed_date', 'count_after'])\n",
    "    \n",
    "    w_h1_df1= Window.partitionBy(h1_df1['ddos_date']).orderBy(F.desc('parsed_date'))\n",
    "    # Create column\n",
    "    h1_df1 = h1_df1.select('*', rank().over(w_h1_df1).alias('ColumnIndex'))\n",
    "    \n",
    "    w_h1_df2= Window.partitionBy(h1_df2['ddos_date']).orderBy(h1_df2['parsed_date'])\n",
    "    # Create column\n",
    "    h1_df2 = h1_df2.select('*', rank().over(w_h1_df2).alias('ColumnIndex'))\n",
    "    \n",
    "    h1_df1_list.append(h1_df1)\n",
    "    h1_df2_list.append(h1_df2)\n",
    "    \n",
    "    \n",
    "hypo1_df1 = reduce(DataFrame.unionAll, h1_df1_list)\n",
    "hypo1_df2 = reduce(DataFrame.unionAll, h1_df2_list)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "h2_df1_list = []\n",
    "h2_df2_list = []\n",
    "\n",
    "for d in ddos_list:\n",
    "    h2_df1, h2_df2 = hypo_2(h2_df, d)\n",
    "    \n",
    "    h2_df1 = h2_df1.dropDuplicates(['parsed_date', 'count_before'])\n",
    "    h2_df2 = h2_df2.dropDuplicates(['parsed_date', 'count_after'])\n",
    "    \n",
    "    w_h2_df1= Window.partitionBy(h2_df1['ddos_date']).orderBy(F.desc('parsed_date'))\n",
    "    # Create column\n",
    "    h2_df1 = h2_df1.select('*', rank().over(w_h2_df1).alias('ColumnIndex'))\n",
    "    \n",
    "    w_h2_df2= Window.partitionBy(h2_df2['ddos_date']).orderBy(h2_df2['parsed_date'])\n",
    "    # Create column\n",
    "    h2_df2 = h2_df2.select('*', rank().over(w_h2_df2).alias('ColumnIndex'))\n",
    "    \n",
    "    h2_df1_list.append(h2_df1)\n",
    "    h2_df2_list.append(h2_df2)\n",
    "    \n",
    "    \n",
    "hypo2_df1 = reduce(DataFrame.unionAll, h2_df1_list)\n",
    "hypo2_df2 = reduce(DataFrame.unionAll, h2_df2_list)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "h3_df1_list = []\n",
    "h3_df2_list = []\n",
    "\n",
    "for d in ddos_list:\n",
    "    h3_df1, h3_df2 = hypo_3(h3_df, d)\n",
    "    \n",
    "    h3_df1 = h3_df1.dropDuplicates(['parsed_date', 'count_before'])\n",
    "    h3_df2 = h3_df2.dropDuplicates(['parsed_date', 'count_after'])\n",
    "    \n",
    "    w_h3_df1= Window.partitionBy(h3_df1['ddos_date']).orderBy(F.desc('parsed_date'))\n",
    "    # Create column\n",
    "    h3_df1 = h3_df1.select('*', rank().over(w_h3_df1).alias('ColumnIndex'))\n",
    "    \n",
    "    w_h3_df2= Window.partitionBy(h3_df2['ddos_date']).orderBy(h3_df2['parsed_date'])\n",
    "    # Create column\n",
    "    h3_df2 = h3_df2.select('*', rank().over(w_h3_df2).alias('ColumnIndex'))\n",
    "    \n",
    "    h3_df1_list.append(h3_df1)\n",
    "    h3_df2_list.append(h3_df2)\n",
    "    \n",
    "\n",
    "hypo3_df1 = reduce(DataFrame.unionAll, h3_df1_list)\n",
    "hypo3_df2 = reduce(DataFrame.unionAll, h3_df2_list)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------------------------------------------------------------------------------------------+-------------------+\n",
      "|ddos_date |arr_h1                                                                                          |u_value & p_value  |\n",
      "+----------+------------------------------------------------------------------------------------------------+-------------------+\n",
      "|2018-01-17|[[1151, 1041], [1074, 1065], [1121, 1306], [767, 1016], [1054, 741], [1277, 1134], [1328, 1194]]|[30.0, 0.52290326] |\n",
      "|2018-01-27|[[1105, 1309], [1194, 1270], [1041, 1066], [1016, 880], [741, 1111], [1100, 795], [1306, 1323]] |[19.0, 0.52290326] |\n",
      "|2018-02-08|[[1143, 1063], [880, 1323], [1231, 910], [1111, 957], [806, 1168], [1052, 833], [1066, 1171]]   |[22.0, 0.7982978]  |\n",
      "|2018-03-01|[[899, 1264], [1155, 925], [1110, 1258], [1023, 1160], [1244, 1426], [1168, 723], [1236, 1280]] |[16.0, 0.30668506] |\n",
      "|2018-03-06|[[1264, 1160], [723, 1280]]                                                                     |[1.0, 0.6985354]   |\n",
      "|2018-05-14|[[1173, 1061], [1151, 1195], [702, 1207], [724, 1395], [1173, 923], [1109, 1231], [1058, 1524]] |[9.0, 0.055017326] |\n",
      "|2018-05-24|[[1195, 1234], [1286, 883], [1061, 1333], [1386, 1195], [923, 1133], [1231, 1439], [1524, 1322]]|[24.5, 0.9490037]  |\n",
      "|2018-07-30|[[1285, 891], [1353, 1374], [1535, 1315], [765, 1195], [1866, 920], [1001, 1344], [1262, 1369]] |[27.0, 0.7982978]  |\n",
      "|2019-01-16|[[1298, 1845]]                                                                                  |[0.0, 1.0]         |\n",
      "|2019-02-23|[[1678, 1483]]                                                                                  |[1.0, 1.0]         |\n",
      "|2019-03-22|[[3052, 9620], [1771, 4478]]                                                                    |[0.0, 0.24527812]  |\n",
      "|2019-04-16|[[1427, 2145], [2026, 1870]]                                                                    |[1.0, 0.6985354]   |\n",
      "|2019-08-08|[[1, 1], [2301, 1], [2092, 1], [2, 1959], [1289, 1]]                                            |[20.0, 0.118797496]|\n",
      "+----------+------------------------------------------------------------------------------------------------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result_1 = hypo1_df1.join(\n",
    "    hypo1_df2, ['ddos_date', 'ColumnIndex']\n",
    ").drop('ColumnIndex').groupBy('ddos_date').agg(\n",
    "    F.collect_list(F.array('count_before', 'count_after')).alias('arr_h1')\n",
    ").withColumn(\n",
    "    'u_value & p_value', \n",
    "    F.udf(\n",
    "        lambda arr:\n",
    "            [float(j) for j in mannwhitneyu([i[0] for i in arr], [i[1] for i in arr], alternative='two-sided')],\n",
    "        'array<float>'\n",
    "    )('arr_h1')\n",
    ")\n",
    "\n",
    "result_1.orderBy(\"ddos_date\").show(30, truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-------------------+\n",
      "|ddos_date |arr_h2                                                                                                                                                                              |u_value & p_value  |\n",
      "+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-------------------+\n",
      "|2018-01-17|[[445, 414], [420, 456], [363, 326], [498, 400], [408, 526], [440, 476], [281, 464], [388, 434], [382, 490], [335, 414], [491, 462], [380, 397], [357, 295], [270, 361], [585, 431]]|[89.0, 0.8402739]  |\n",
      "|2018-01-27|[[464, 315], [431, 490], [295, 502], [462, 414], [456, 326], [397, 317], [380, 440], [498, 325], [414, 466], [281, 424], [400, 434], [476, 526], [357, 367], [445, 497], [468, 487]]|[100.5, 0.697957]  |\n",
      "|2018-02-08|[[502, 406], [497, 424], [476, 357], [361, 481], [431, 487], [317, 456], [456, 433], [487, 367], [434, 428], [526, 321], [315, 565], [326, 463], [490, 380], [466, 325], [414, 443]]|[132.0, 0.21527271]|\n",
      "|2018-03-01|[[344, 502], [465, 460], [411, 257], [281, 476], [357, 497], [487, 517], [428, 332]]                                                                                                |[16.0, 0.87492347] |\n",
      "|2018-03-06|[[476, 497], [257, 517]]                                                                                                                                                            |[0.0, 0.9735962]   |\n",
      "|2018-05-14|[[528, 343], [413, 619], [444, 421], [373, 457], [205, 497], [211, 443], [260, 684], [292, 572], [566, 430], [398, 342], [375, 565], [313, 505], [389, 475], [437, 443], [236, 507]]|[43.0, 0.9981566]  |\n",
      "|2018-05-24|[[443, 457], [528, 530], [437, 470], [430, 421], [343, 572], [507, 582], [443, 342], [353, 395], [497, 345], [236, 538], [389, 468], [565, 619], [205, 556], [475, 649], [684, 457]]|[79.0, 0.920814]   |\n",
      "|2018-07-30|[[528, 580], [481, 343], [537, 486], [509, 320], [324, 538], [398, 591], [551, 521], [264, 534], [400, 509], [457, 546], [501, 322], [586, 519], [781, 421], [671, 322], [361, 497]]|[117.5, 0.42595074]|\n",
      "|2019-01-16|[[487, 801]]                                                                                                                                                                        |[0.0, 0.97724986]  |\n",
      "|2019-02-23|[[629, 624]]                                                                                                                                                                        |[1.0, 0.5]         |\n",
      "|2019-03-22|[[1240, 4293], [724, 2428]]                                                                                                                                                         |[0.0, 0.9735962]   |\n",
      "|2019-04-16|[[457, 896], [723, 717]]                                                                                                                                                            |[1.0, 0.87736094]  |\n",
      "|2019-08-08|[[925, 772], [857, 1]]                                                                                                                                                              |[4.0, 0.12263906]  |\n",
      "+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result_2 = hypo2_df1.join(\n",
    "    hypo2_df2, ['ddos_date', 'ColumnIndex']\n",
    ").drop('ColumnIndex').groupBy('ddos_date').agg(\n",
    "    F.collect_list(F.array('count_before', 'count_after')).alias('arr_h2')\n",
    ").withColumn(\n",
    "    'u_value & p_value', \n",
    "    F.udf(\n",
    "        lambda arr:\n",
    "            [float(j) for j in mannwhitneyu([i[0] for i in arr], [i[1] for i in arr], alternative='greater')],\n",
    "        'array<float>'\n",
    "    )('arr_h2')\n",
    ")\n",
    "\n",
    "result_2.orderBy(\"ddos_date\").show(25, truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-------------------+\n",
      "|ddos_date |arr_h3                                                                                                                                                                         |u_value & p_value  |\n",
      "+----------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-------------------+\n",
      "|2018-01-17|[[221, 105], [211, 180], [65, 152], [70, 139], [155, 132], [108, 111], [161, 137], [153, 230], [76, 155], [115, 91], [117, 174], [154, 241], [108, 238], [127, 81], [255, 102]]|[99.5, 0.3020229]  |\n",
      "|2018-01-27|[[117, 226], [108, 95], [221, 141], [180, 155], [105, 105], [65, 153], [174, 83], [127, 116], [152, 97], [81, 238], [238, 241], [196, 258], [102, 230], [132, 91], [111, 137]] |[97.0, 0.2668692]  |\n",
      "|2018-02-08|[[258, 116], [241, 86], [155, 99], [238, 129], [141, 153], [230, 81], [132, 133], [137, 167], [139, 161], [97, 170], [91, 164], [83, 163], [238, 111], [105, 95], [102, 186]]  |[129.0, 0.7596589] |\n",
      "|2018-03-01|[[91, 133], [48, 253], [111, 168], [131, 349], [122, 58], [148, 95], [186, 213]]                                                                                               |[14.0, 0.100668244]|\n",
      "|2018-03-06|[[253, 168], [58, 213]]                                                                                                                                                        |[2.0, 0.65073234]  |\n",
      "|2018-05-14|[[172, 551], [100, 134], [83, 142], [264, 194], [127, 67], [216, 74], [129, 118], [121, 65], [34, 112], [103, 114], [148, 121], [62, 85], [53, 310], [130, 32], [89, 127]]     |[110.0, 0.46693587]|\n",
      "|2018-05-24|[[34, 181], [129, 130], [194, 74], [114, 551], [118, 132], [85, 32], [142, 119], [97, 94], [121, 219], [112, 100], [83, 217], [127, 159], [148, 159], [310, 65], [67, 134]]    |[90.0, 0.18072148] |\n",
      "|2018-07-30|[[107, 177], [165, 175], [234, 156], [155, 60], [110, 199], [155, 63], [155, 137], [181, 115], [82, 157], [227, 90], [104, 120], [124, 198], [172, 60], [134, 148], [363, 108]]|[137.0, 0.8502553] |\n",
      "|2019-01-16|[[119, 285]]                                                                                                                                                                   |[0.0, 0.5]         |\n",
      "|2019-02-23|[[273, 186]]                                                                                                                                                                   |[1.0, 0.97724986]  |\n",
      "|2019-03-22|[[198, 679], [210, 264]]                                                                                                                                                       |[0.0, 0.12263906]  |\n",
      "|2019-04-16|[[127, 254], [268, 268]]                                                                                                                                                       |[1.5, 0.5]         |\n",
      "|2019-08-08|[[234, 108]]                                                                                                                                                                   |[1.0, 0.97724986]  |\n",
      "+----------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result_3 = hypo3_df1.join(\n",
    "    hypo3_df2, ['ddos_date', 'ColumnIndex']\n",
    ").drop('ColumnIndex').groupBy('ddos_date').agg(\n",
    "    F.collect_list(F.array('count_before', 'count_after')).alias('arr_h3')\n",
    ").withColumn(\n",
    "    'u_value & p_value', \n",
    "    F.udf(\n",
    "        lambda arr:\n",
    "            [float(j) for j in mannwhitneyu([i[0] for i in arr], [i[1] for i in arr], alternative='less')],\n",
    "        'array<float>'\n",
    "    )('arr_h3')\n",
    ")\n",
    "\n",
    "result_3.orderBy(\"ddos_date\").show(25, truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Stop the SparkContext\n",
    "note: don't run this block unless you actually want to stop your context"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2021-05-06 00:38:25] Stopping Spark context.\n"
     ]
    }
   ],
   "source": [
    "print(\"[{}] Stopping Spark context.\".format(datetime.now().replace(microsecond=0)))\n",
    "sc.stop()"
   ]
  }
 ],
 "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}