hands_on_pyspark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "\n",
    "from pyspark import SparkContext\n",
    "sc = SparkContext('local', 'Hands on PySpark')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# sc.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[3650, 1095, 12775, 9125, 14965, 3285, 10585]\n"
     ]
    }
   ],
   "source": [
    "visitors = [10, 3, 35, 25, 41, 9, 29]\n",
    "df_visitors = sc.parallelize(visitors)\n",
    "df_visitors_yearly = df_visitors.map(lambda x: x*365).collect()\n",
    "print(df_visitors_yearly)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[10, 3, 35]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_visitors.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[3650, 1095, 12775, 9125, 14965, 3285, 10585]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_visitors_yearly"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "import urllib.request\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# try:\n",
    "#     urllib.request.urlretrieve(\"https://archive.ics.uci.edu/ml/machine-learning-databases/kddcup99-mld/kddcup.data.gz\"),\"kddcup.data.gz\"\n",
    "# except Exception as e:\n",
    "#     print(\"Could not download data\", e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# dd  =  urllib.request.urlretrieve(\"https://archive.ics.uci.edu/ml/machine-learning-databases/kddcup99-mld/kddcup.data.gz\"),\"kddcup.data.gz\"\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--2020-01-17 18:26:18--  https://archive.ics.uci.edu/ml/machine-learning-databases/kddcup99-mld/kddcup.data.gz\n",
      "Resolving archive.ics.uci.edu (archive.ics.uci.edu)... 128.195.10.252\n",
      "Connecting to archive.ics.uci.edu (archive.ics.uci.edu)|128.195.10.252|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 18115902 (17M) [application/x-httpd-php]\n",
      "Saving to: ‘kddcup.data.gz.1’\n",
      "\n",
      "kddcup.data.gz.1    100%[===================>]  17.28M  1.56MB/s    in 17s     \n",
      "\n",
      "2020-01-17 18:26:36 (1.01 MB/s) - ‘kddcup.data.gz.1’ saved [18115902/18115902]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "!wget https://archive.ics.uci.edu/ml/machine-learning-databases/kddcup99-mld/kddcup.data.gz"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://192.168.86.236:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.4</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>Hands on PySpark</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local appName=Hands on PySpark>"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_data = sc.textFile(\"kddcup.data.gz\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "kddcup.data.gz MapPartitionsRDD[4] at textFile at NativeMethodAccessorImpl.java:0"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "raw_data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "total 73880\r\n",
      "-rw-r--r--   1 director  staff  18115902 26 Jun  2007 kddcup.data.gz\r\n",
      "-rw-r--r--   1 director  staff  18115902 26 Jun  2007 kddcup.data.gz.1\r\n",
      "-rw-r--r--   1 director  staff      3913 17 Jan 18:26 hands_on_pyspark.ipynb\r\n",
      "-rw-r--r--   1 director  staff      1648  6 Feb  2019 GraphFrames_Pyspark.ipynb\r\n",
      "-rw-r--r--@  1 director  staff      1304  6 Feb  2019 streams.py\r\n",
      "-rw-r--r--@  1 director  staff       963  6 Feb  2019 streaming_kafka.py\r\n",
      "drwxr-xr-x  20 director  staff       640 23 Dec 14:23 \u001b[1m\u001b[36m..\u001b[m\u001b[m\r\n",
      "drwxr-xr-x   9 director  staff       288 17 Jan 18:26 \u001b[1m\u001b[36m.\u001b[m\u001b[m\r\n",
      "drwxr-xr-x   4 director  staff       128 17 Jan 17:40 \u001b[1m\u001b[36m.ipynb_checkpoints\u001b[m\u001b[m\r\n"
     ]
    }
   ],
   "source": [
    "!ls -laS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "split_file = raw_data.map(lambda line: line.split(\",\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['0',\n",
       "  'tcp',\n",
       "  'http',\n",
       "  'SF',\n",
       "  '215',\n",
       "  '45076',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '1',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '1',\n",
       "  '1',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '1.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  'normal.'],\n",
       " ['0',\n",
       "  'tcp',\n",
       "  'http',\n",
       "  'SF',\n",
       "  '162',\n",
       "  '4528',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '1',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '2',\n",
       "  '2',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '1.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '1',\n",
       "  '1',\n",
       "  '1.00',\n",
       "  '0.00',\n",
       "  '1.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  'normal.'],\n",
       " ['0',\n",
       "  'tcp',\n",
       "  'http',\n",
       "  'SF',\n",
       "  '236',\n",
       "  '1228',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '1',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '1',\n",
       "  '1',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '1.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '2',\n",
       "  '2',\n",
       "  '1.00',\n",
       "  '0.00',\n",
       "  '0.50',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  'normal.'],\n",
       " ['0',\n",
       "  'tcp',\n",
       "  'http',\n",
       "  'SF',\n",
       "  '233',\n",
       "  '2032',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '1',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '2',\n",
       "  '2',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '1.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '3',\n",
       "  '3',\n",
       "  '1.00',\n",
       "  '0.00',\n",
       "  '0.33',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  'normal.'],\n",
       " ['0',\n",
       "  'tcp',\n",
       "  'http',\n",
       "  'SF',\n",
       "  '239',\n",
       "  '486',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '1',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '3',\n",
       "  '3',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '1.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '4',\n",
       "  '4',\n",
       "  '1.00',\n",
       "  '0.00',\n",
       "  '0.25',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  'normal.'],\n",
       " ['0',\n",
       "  'tcp',\n",
       "  'http',\n",
       "  'SF',\n",
       "  '238',\n",
       "  '1282',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '1',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '0',\n",
       "  '4',\n",
       "  '4',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '1.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '5',\n",
       "  '5',\n",
       "  '1.00',\n",
       "  '0.00',\n",
       "  '0.20',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  '0.00',\n",
       "  'normal.']]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "split_file.take(6)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# raw_data.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "from time import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "sampled = raw_data.sample(False, 0.1, 42)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "contains_normal_sample = sampled.map(lambda x: x.split(\",\")).filter(lambda x: \"normal\" in x)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "t0 = time()\n",
    "num_sampled = contains_normal_sample.count()\n",
    "duration = time() - t0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "29.59354019165039"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "duration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "contains_normal = raw_data.map(lambda x: x.split(\",\")).filter(lambda x: \"normal\" in x)\n",
    "t0 = time()\n",
    "num_sampled = contains_normal.count()\n",
    "duration = time() - t0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "53.96395492553711"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "duration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_in_memory = raw_data.takeSample(False, 10, 42)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "contains_normal_py = [line.split(\",\") for line in data_in_memory if \"normal\" in line]\n",
    "len(contains_normal_py)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "normal_sample = sampled.filter(lambda line: \"normal.\" in line)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "non_normal_sample = sampled.subtract(normal_sample)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "490705"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sampled.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "97404"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "normal_sample.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "393301"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "non_normal_sample.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "feature_1 = sampled.map(lambda line: line.split(\",\")).map(lambda features: features[1]).distinct()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "feature_2 = sampled.map(lambda line: line.split(\",\")).map(lambda features: features[2]).distinct()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "f1 = feature_1.collect()\n",
    "f2 = feature_2.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['tcp', 'udp', 'icmp']"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "f1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['http',\n",
       " 'finger',\n",
       " 'auth',\n",
       " 'domain_u',\n",
       " 'smtp',\n",
       " 'ftp',\n",
       " 'telnet',\n",
       " 'eco_i',\n",
       " 'ntp_u',\n",
       " 'ecr_i',\n",
       " 'other',\n",
       " 'private',\n",
       " 'pop_3',\n",
       " 'ftp_data',\n",
       " 'daytime',\n",
       " 'remote_job',\n",
       " 'supdup',\n",
       " 'name',\n",
       " 'ssh',\n",
       " 'domain',\n",
       " 'gopher',\n",
       " 'time',\n",
       " 'rje',\n",
       " 'ctf',\n",
       " 'mtp',\n",
       " 'X11',\n",
       " 'urp_i',\n",
       " 'pm_dump',\n",
       " 'IRC',\n",
       " 'exec',\n",
       " 'bgp',\n",
       " 'nnsp',\n",
       " 'iso_tsap',\n",
       " 'http_443',\n",
       " 'login',\n",
       " 'shell',\n",
       " 'printer',\n",
       " 'efs',\n",
       " 'courier',\n",
       " 'uucp',\n",
       " 'kshell',\n",
       " 'klogin',\n",
       " 'whois',\n",
       " 'echo',\n",
       " 'discard',\n",
       " 'systat',\n",
       " 'netstat',\n",
       " 'hostnames',\n",
       " 'csnet_ns',\n",
       " 'pop_2',\n",
       " 'sunrpc',\n",
       " 'uucp_path',\n",
       " 'nntp',\n",
       " 'netbios_ns',\n",
       " 'netbios_ssn',\n",
       " 'netbios_dgm',\n",
       " 'imap4',\n",
       " 'sql_net',\n",
       " 'vmnet',\n",
       " 'link',\n",
       " 'Z39_50',\n",
       " 'ldap',\n",
       " 'urh_i',\n",
       " 'tftp_u',\n",
       " 'red_i',\n",
       " 'tim_i']"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "f2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "198"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# all the combination between f1 and f2\n",
    "len(feature_1.cartesian(feature_2).collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [],
   "source": [
    "csv = raw_data.map(lambda x: x.split(\",\"))\n",
    "normal_data = csv.filter(lambda x: x[41]==\"normal.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "211895753"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Get total duration\n",
    "duration = normal_data.map(lambda x: int(x[0]))\n",
    "total_duration = duration.reduce(lambda x, y: x+y)\n",
    "total_duration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "217.82472416710442"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# average duration\n",
    "total_duration/(normal_data.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Computations using Aggregate"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [],
   "source": [
    "duration_count = duration.aggregate(\n",
    "    (0,0),\n",
    "    (lambda db, new_value: (db[0] + new_value, db[1] + 1)),\n",
    "    (lambda db1, db2: (db1[0] + db2[0], db1[1] + db2[1]))\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "217.82472416710442"
      ]
     },
     "execution_count": 49,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "duration_count[0]/duration_count[1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('normal.', 211895753.0),\n",
       " ('buffer_overflow.', 2751.0),\n",
       " ('loadmodule.', 326.0),\n",
       " ('perl.', 124.0),\n",
       " ('neptune.', 2.0),\n",
       " ('smurf.', 0.0),\n",
       " ('guess_passwd.', 144.0),\n",
       " ('pod.', 0.0),\n",
       " ('teardrop.', 0.0),\n",
       " ('portsweep.', 24257982.0),\n",
       " ('ipsweep.', 13049.0),\n",
       " ('land.', 0.0),\n",
       " ('ftp_write.', 259.0),\n",
       " ('back.', 284.0),\n",
       " ('imap.', 72.0),\n",
       " ('satan.', 500.0),\n",
       " ('phf.', 18.0),\n",
       " ('nmap.', 0.0),\n",
       " ('multihop.', 1288.0),\n",
       " ('warezmaster.', 301.0),\n",
       " ('warezclient.', 627563.0),\n",
       " ('spy.', 636.0),\n",
       " ('rootkit.', 1008.0)]"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# PIVOT\n",
    "kv_duration = csv.map(lambda x: (x[41], float(x[0]))).reduceByKey(lambda x, y: x+y)\n",
    "kv_duration.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('normal.',\n",
       "  ['0',\n",
       "   'tcp',\n",
       "   'http',\n",
       "   'SF',\n",
       "   '215',\n",
       "   '45076',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '1',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '1',\n",
       "   '1',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '1.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   'normal.'])]"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kv = csv.map(lambda x: (x[41], x))\n",
    "kv.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "defaultdict(int,\n",
       "            {'normal.': 972781,\n",
       "             'buffer_overflow.': 30,\n",
       "             'loadmodule.': 9,\n",
       "             'perl.': 3,\n",
       "             'neptune.': 1072017,\n",
       "             'smurf.': 2807886,\n",
       "             'guess_passwd.': 53,\n",
       "             'pod.': 264,\n",
       "             'teardrop.': 979,\n",
       "             'portsweep.': 10413,\n",
       "             'ipsweep.': 12481,\n",
       "             'land.': 21,\n",
       "             'ftp_write.': 8,\n",
       "             'back.': 2203,\n",
       "             'imap.': 12,\n",
       "             'satan.': 15892,\n",
       "             'phf.': 4,\n",
       "             'nmap.': 2316,\n",
       "             'multihop.': 7,\n",
       "             'warezmaster.': 20,\n",
       "             'warezclient.': 1020,\n",
       "             'spy.': 2,\n",
       "             'rootkit.': 10})"
      ]
     },
     "execution_count": 52,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kv.countByKey()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}