{
 "metadata": {
  "name": "",
  "signature": "sha256:01c23f9757cbbcc111ac851dc4f48c2a8732f1271f4cd54e1f91839a9d62a8dc"
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "Working with key/value pair RDDs"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "[Introduction to Spark with Python, by Jose A. Dianes](https://github.com/jadianes/spark-py-notebooks)"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Spark provides specific functions to deal with RDDs which elements are key/value pairs. They are usually used to perform aggregations and other processings by key.  "
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "In this notebook we will show how, by working with key/value pairs, we can process our network interactions dataset in a more practical and powerful way than that used in previous notebooks. Key/value pair aggregations will show to be particularly effective when trying to explore each type of tag in our network attacks, in an individual way.  "
     ]
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Getting the data and creating the RDD"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "As we did in our first notebook, we will use the reduced dataset (10 percent) provided for the [KDD Cup 1999](http://kdd.ics.uci.edu/databases/kddcup99/kddcup99.html), containing nearly half million network interactions. The file is provided as a Gzip file that we will download locally.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import urllib\n",
      "f = urllib.urlretrieve (\"http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz\", \"kddcup.data_10_percent.gz\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 1
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "data_file = \"./kddcup.data_10_percent.gz\"\n",
      "raw_data = sc.textFile(data_file)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 1
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Creating a pair RDD for interaction types"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "In this notebook we want to do some exploratory data analysis on our network interactions dataset. More concretely we want to profile each network interaction type in terms of some of its variables such as duration. In order to do so, we first need to create the RDD suitable for that, where each interaction is parsed as a CSV row representing the value, and is put together with its corresponding tag as a key.  "
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Normally we create key/value pair RDDs by applying a function using `map` to the original data. This function returns the corresponding pair for a given RDD element. We can proceed as follows.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "csv_data = raw_data.map(lambda x: x.split(\",\"))\n",
      "key_value_data = csv_data.map(lambda x: (x[41], x)) # x[41] contains the network interaction tag"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 2
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "We have now our key/value pair data ready to be used. Let's get the first element in order to see how it looks like.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "key_value_data.take(1)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 3,
       "text": [
        "[(u'normal.',\n",
        "  [u'0',\n",
        "   u'tcp',\n",
        "   u'http',\n",
        "   u'SF',\n",
        "   u'181',\n",
        "   u'5450',\n",
        "   u'0',\n",
        "   u'0',\n",
        "   u'0',\n",
        "   u'0',\n",
        "   u'0',\n",
        "   u'1',\n",
        "   u'0',\n",
        "   u'0',\n",
        "   u'0',\n",
        "   u'0',\n",
        "   u'0',\n",
        "   u'0',\n",
        "   u'0',\n",
        "   u'0',\n",
        "   u'0',\n",
        "   u'0',\n",
        "   u'8',\n",
        "   u'8',\n",
        "   u'0.00',\n",
        "   u'0.00',\n",
        "   u'0.00',\n",
        "   u'0.00',\n",
        "   u'1.00',\n",
        "   u'0.00',\n",
        "   u'0.00',\n",
        "   u'9',\n",
        "   u'9',\n",
        "   u'1.00',\n",
        "   u'0.00',\n",
        "   u'0.11',\n",
        "   u'0.00',\n",
        "   u'0.00',\n",
        "   u'0.00',\n",
        "   u'0.00',\n",
        "   u'0.00',\n",
        "   u'normal.'])]"
       ]
      }
     ],
     "prompt_number": 3
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Data aggregations with key/value pair RDDs"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "We can use all the transformations and actions available for normal RDDs with key/value pair RDDs. We just need to make the functions work with pair elements. Additionally, Spark provides specific functions to work with RDDs containing pair elements. They are very similar to those available for general RDDs.  "
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "For example, we have a `reduceByKey` transformation that we can use as follows to calculate the total duration of each network interaction type.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "key_value_duration = csv_data.map(lambda x: (x[41], float(x[0]))) \n",
      "durations_by_key = key_value_duration.reduceByKey(lambda x, y: x + y)\n",
      "\n",
      "durations_by_key.collect()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 4,
       "text": [
        "[(u'guess_passwd.', 144.0),\n",
        " (u'nmap.', 0.0),\n",
        " (u'warezmaster.', 301.0),\n",
        " (u'rootkit.', 1008.0),\n",
        " (u'warezclient.', 627563.0),\n",
        " (u'smurf.', 0.0),\n",
        " (u'pod.', 0.0),\n",
        " (u'neptune.', 0.0),\n",
        " (u'normal.', 21075991.0),\n",
        " (u'spy.', 636.0),\n",
        " (u'ftp_write.', 259.0),\n",
        " (u'phf.', 18.0),\n",
        " (u'portsweep.', 1991911.0),\n",
        " (u'teardrop.', 0.0),\n",
        " (u'buffer_overflow.', 2751.0),\n",
        " (u'land.', 0.0),\n",
        " (u'imap.', 72.0),\n",
        " (u'loadmodule.', 326.0),\n",
        " (u'perl.', 124.0),\n",
        " (u'multihop.', 1288.0),\n",
        " (u'back.', 284.0),\n",
        " (u'ipsweep.', 43.0),\n",
        " (u'satan.', 64.0)]"
       ]
      }
     ],
     "prompt_number": 4
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "We have a specific counting action for key/value pairs.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "counts_by_key = key_value_data.countByKey()\n",
      "counts_by_key"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 5,
       "text": [
        "defaultdict(<type 'int'>, {u'guess_passwd.': 53, u'nmap.': 231, u'warezmaster.': 20, u'rootkit.': 10, u'warezclient.': 1020, u'smurf.': 280790, u'pod.': 264, u'neptune.': 107201, u'normal.': 97278, u'spy.': 2, u'ftp_write.': 8, u'phf.': 4, u'portsweep.': 1040, u'teardrop.': 979, u'buffer_overflow.': 30, u'land.': 21, u'imap.': 12, u'loadmodule.': 9, u'perl.': 3, u'multihop.': 7, u'back.': 2203, u'ipsweep.': 1247, u'satan.': 1589})"
       ]
      }
     ],
     "prompt_number": 5
    },
    {
     "cell_type": "heading",
     "level": 3,
     "metadata": {},
     "source": [
      "Using `combineByKey`"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "This is the most general of the per-key aggregation functions. Most of the other per-key combiners are implemented using it. We can think about it as the `aggregate` equivalent since it allows the user to return values that are not the same type as our input data."
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "For example, we can use it to calculate per-type average durations as follows.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "sum_counts = key_value_duration.combineByKey(\n",
      "    (lambda x: (x, 1)), # the initial value, with value x and count 1\n",
      "    (lambda acc, value: (acc[0]+value, acc[1]+1)), # how to combine a pair value with the accumulator: sum value, and increment count\n",
      "    (lambda acc1, acc2: (acc1[0]+acc2[0], acc1[1]+acc2[1])) # combine accumulators\n",
      ")\n",
      "\n",
      "sum_counts.collectAsMap()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 6,
       "text": [
        "{u'back.': (284.0, 2203),\n",
        " u'buffer_overflow.': (2751.0, 30),\n",
        " u'ftp_write.': (259.0, 8),\n",
        " u'guess_passwd.': (144.0, 53),\n",
        " u'imap.': (72.0, 12),\n",
        " u'ipsweep.': (43.0, 1247),\n",
        " u'land.': (0.0, 21),\n",
        " u'loadmodule.': (326.0, 9),\n",
        " u'multihop.': (1288.0, 7),\n",
        " u'neptune.': (0.0, 107201),\n",
        " u'nmap.': (0.0, 231),\n",
        " u'normal.': (21075991.0, 97278),\n",
        " u'perl.': (124.0, 3),\n",
        " u'phf.': (18.0, 4),\n",
        " u'pod.': (0.0, 264),\n",
        " u'portsweep.': (1991911.0, 1040),\n",
        " u'rootkit.': (1008.0, 10),\n",
        " u'satan.': (64.0, 1589),\n",
        " u'smurf.': (0.0, 280790),\n",
        " u'spy.': (636.0, 2),\n",
        " u'teardrop.': (0.0, 979),\n",
        " u'warezclient.': (627563.0, 1020),\n",
        " u'warezmaster.': (301.0, 20)}"
       ]
      }
     ],
     "prompt_number": 6
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "We can see that the arguments are pretty similar to those passed to `aggregate` in the previous notebook. The result associated to each type is in the form of a pair. If we want to actually get the averages, we need to do the division before collecting the results.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "duration_means_by_type = sum_counts.map(lambda (key,value): (key, round(value[0]/value[1],3))).collectAsMap()\n",
      "\n",
      "# Print them sorted\n",
      "for tag in sorted(duration_means_by_type, key=duration_means_by_type.get, reverse=True):\n",
      "    print tag, duration_means_by_type[tag]"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "portsweep. 1915.299\n",
        "warezclient. 615.258\n",
        "spy. 318.0\n",
        "normal. 216.657\n",
        "multihop. 184.0\n",
        "rootkit. 100.8\n",
        "buffer_overflow. 91.7\n",
        "perl. 41.333\n",
        "loadmodule. 36.222\n",
        "ftp_write. 32.375\n",
        "warezmaster. 15.05\n",
        "imap. 6.0\n",
        "phf. 4.5\n",
        "guess_passwd. 2.717\n",
        "back. 0.129\n",
        "satan. 0.04\n",
        "ipsweep. 0.034\n",
        "nmap. 0.0\n",
        "smurf. 0.0\n",
        "pod. 0.0\n",
        "neptune. 0.0\n",
        "teardrop. 0.0\n",
        "land. 0.0\n"
       ]
      }
     ],
     "prompt_number": 7
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "A small step into understanding what makes a network interaction be considered an attack."
     ]
    }
   ],
   "metadata": {}
  }
 ]
}