spark_tutorial.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# RDD creation"
   ]
  },
  {
   "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": [
    "In this notebook we will introduce two different ways of getting data into the basic Spark data structure, the **Resilient Distributed Dataset** or **RDD**. An RDD is a distributed collection of elements. All work in Spark is expressed as either creating new RDDs, transforming existing RDDs, or calling actions on RDDs to compute a result. Spark automatically distributes the data contained in RDDs across your cluster and parallelizes the operations you perform on them."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### References"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The reference book for these and other Spark related topics is *Learning Spark* by Holden Karau, Andy Konwinski, Patrick Wendell, and Matei Zaharia.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The KDD Cup 1999 competition dataset is described in detail [here](http://kdd.ics.uci.edu/databases/kddcup99/kddcup99)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Getting the data files  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this notebook we will use the reduced dataset (10 percent) provided for the KDD Cup 1999, containing nearly half million network interactions. The file is provided as a *Gzip* file that we will download locally.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import urllib\n",
    "f = urllib.urlretrieve (\"http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz\", \"kddcup.data_10_percent.gz\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating a RDD from a file  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The most common way of creating an RDD is to load it from a file. Notice that Spark's `textFile` can handle compressed files directly.    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "data_file = \"./kddcup.data_10_percent.gz\"\n",
    "raw_data = sc.textFile(data_file)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we have our data file loaded into the `raw_data` RDD."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Without getting into Spark *transformations* and *actions*, the most basic thing we can do to check that we got our RDD contents right is to `count()` the number of lines loaded from the file into the RDD.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "494021"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "raw_data.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also check the first few entries in our data.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'0,tcp,http,SF,181,5450,0,0,0,0,0,1,0,0,0,0,0,0,0,0,0,0,8,8,0.00,0.00,0.00,0.00,1.00,0.00,0.00,9,9,1.00,0.00,0.11,0.00,0.00,0.00,0.00,0.00,normal.',\n",
       " u'0,tcp,http,SF,239,486,0,0,0,0,0,1,0,0,0,0,0,0,0,0,0,0,8,8,0.00,0.00,0.00,0.00,1.00,0.00,0.00,19,19,1.00,0.00,0.05,0.00,0.00,0.00,0.00,0.00,normal.',\n",
       " u'0,tcp,http,SF,235,1337,0,0,0,0,0,1,0,0,0,0,0,0,0,0,0,0,8,8,0.00,0.00,0.00,0.00,1.00,0.00,0.00,29,29,1.00,0.00,0.03,0.00,0.00,0.00,0.00,0.00,normal.',\n",
       " u'0,tcp,http,SF,219,1337,0,0,0,0,0,1,0,0,0,0,0,0,0,0,0,0,6,6,0.00,0.00,0.00,0.00,1.00,0.00,0.00,39,39,1.00,0.00,0.03,0.00,0.00,0.00,0.00,0.00,normal.',\n",
       " u'0,tcp,http,SF,217,2032,0,0,0,0,0,1,0,0,0,0,0,0,0,0,0,0,6,6,0.00,0.00,0.00,0.00,1.00,0.00,0.00,49,49,1.00,0.00,0.02,0.00,0.00,0.00,0.00,0.00,normal.']"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "raw_data.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the following notebooks, we will use this raw data to learn about the different Spark transformations and actions.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating and RDD using `parallelize`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Another way of creating an RDD is to parallelize an already existing list.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "a = range(100)\n",
    "\n",
    "data = sc.parallelize(a)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we did before, we can `count()` the number of elements in the RDD."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "100"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As before, we can access the first few elements on our RDD.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0, 1, 2, 3, 4]"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# RDD basics"
   ]
  },
  {
   "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": [
    "This notebook will introduce three basic but essential Spark operations. Two of them are the *transformations* `map` and `filter`. The other is the *action* `collect`. At the same time we will introduce the concept of *persistence* in Spark.    "
   ]
  },
  {
   "cell_type": "markdown",
   "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, containing nearly half million network interactions. The file is provided as a Gzip file that we will download locally."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import urllib\n",
    "f = urllib.urlretrieve (\"http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz\", \"kddcup.data_10_percent.gz\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can use this file to create our RDD."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "data_file = \"./kddcup.data_10_percent.gz\"\n",
    "raw_data = sc.textFile(data_file)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## The `filter` transformation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This transformation can be applied to RDDs in order to keep just elements that satisfy a certain condition. More concretely, a function is evaluated on every element in the original RDD. The new resulting RDD will contain just those elements that make the function return `True`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For example, imagine we want to count how many `normal.` interactions we have in our dataset. We can filter our `raw_data` RDD as follows.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "normal_raw_data = raw_data.filter(lambda x: 'normal.' in x)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can count how many elements we have in the new RDD."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "There are 97278 'normal' interactions\n",
      "Count completed in 5.951 seconds\n"
     ]
    }
   ],
   "source": [
    "from time import time\n",
    "t0 = time()\n",
    "normal_count = normal_raw_data.count()\n",
    "tt = time() - t0\n",
    "print \"There are {} 'normal' interactions\".format(normal_count)\n",
    "print \"Count completed in {} seconds\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Remember from notebook 1 that we have a total of 494021 in our 10 percent dataset. Here we can see that 97278 contain the `normal.` tag word.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice that we have measured the elapsed time for counting the elements in the RDD. We have done this because we wanted to point out that actual (distributed) computations in Spark take place when we execute *actions* and not *transformations*. In this case `count` is the action we execute on the RDD. We can apply as many transformations as we want on a our RDD and no computation will take place until we call the first action that, in this case takes a few seconds to complete."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## The `map` transformation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "By using the `map` transformation in Spark, we can apply a function to every element in our RDD. Python's lambdas are specially expressive for this particular."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this case we want to read our data file as a CSV formatted one. We can do this by applying a lambda function to each element in the RDD as follows."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Parse completed in 1.715 seconds\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.']\n"
     ]
    }
   ],
   "source": [
    "from pprint import pprint\n",
    "csv_data = raw_data.map(lambda x: x.split(\",\"))\n",
    "t0 = time()\n",
    "head_rows = csv_data.take(5)\n",
    "tt = time() - t0\n",
    "print \"Parse completed in {} seconds\".format(round(tt,3))\n",
    "pprint(head_rows[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Again, all action happens once we call the first Spark *action* (i.e. *take* in this case). What if we take a lot of elements instead of just the first few?  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Parse completed in 8.629 seconds\n"
     ]
    }
   ],
   "source": [
    "t0 = time()\n",
    "head_rows = csv_data.take(100000)\n",
    "tt = time() - t0\n",
    "print \"Parse completed in {} seconds\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can see that it takes longer. The `map` function is applied now in a  distributed way to a lot of elements on the RDD, hence the longer execution time."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using `map` and predefined functions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Of course we can use predefined functions with `map`. Imagine we want to have each element in the RDD as a key-value pair where the key is the tag (e.g. *normal*) and the value is the whole list of elements that represents the row in the CSV formatted file. We could proceed as follows.    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "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.'])\n"
     ]
    }
   ],
   "source": [
    "def parse_interaction(line):\n",
    "    elems = line.split(\",\")\n",
    "    tag = elems[41]\n",
    "    return (tag, elems)\n",
    "\n",
    "key_csv_data = raw_data.map(parse_interaction)\n",
    "head_rows = key_csv_data.take(5)\n",
    "pprint(head_rows[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "That was easy, wasn't it?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In our notebook about working with key-value pairs we will use this type of RDDs to do data aggregations (e.g. count by key)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## The `collect` action"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So far we have used the actions `count` and `take`. Another basic action we need to learn is `collect`. Basically it will get all the elements in the RDD into memory for us to work with them. For this reason it has to be used with care, specially when working with large RDDs.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "An example using our raw data.    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data collected in 17.927 seconds\n"
     ]
    }
   ],
   "source": [
    "t0 = time()\n",
    "all_raw_data = raw_data.collect()\n",
    "tt = time() - t0\n",
    "print \"Data collected in {} seconds\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "That took longer as any other action we used before, of course. Every Spark worker node that has a fragment of the RDD has to be coordinated in order to retrieve its part, and then *reduce* everything together.    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As a last example combining all the previous, we want to collect all the `normal` interactions as key-value pairs.   "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data collected in 12.485 seconds\n",
      "There are 97278 normal interactions\n"
     ]
    }
   ],
   "source": [
    "# get data from file\n",
    "data_file = \"./kddcup.data_10_percent.gz\"\n",
    "raw_data = sc.textFile(data_file)\n",
    "\n",
    "# parse into key-value pairs\n",
    "key_csv_data = raw_data.map(parse_interaction)\n",
    "\n",
    "# filter normal key interactions\n",
    "normal_key_interactions = key_csv_data.filter(lambda x: x[0] == \"normal.\")\n",
    "\n",
    "# collect all\n",
    "t0 = time()\n",
    "all_normal = normal_key_interactions.collect()\n",
    "tt = time() - t0\n",
    "normal_count = len(all_normal)\n",
    "print \"Data collected in {} seconds\".format(round(tt,3))\n",
    "print \"There are {} 'normal' interactions\".format(normal_count)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This count matches with the previous count for `normal` interactions. The new procedure is more time consuming. This is because we retrieve all the data with `collect` and then use Python's `len` on the resulting list. Before we were just counting the total number of elements in the RDD by using `count`.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Sampling 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": [
    "So far we have introduced RDD creation together with some basic transformations such as `map` and `filter` and some actions such as `count`, `take`, and `collect`.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook will show how to sample RDDs. Regarding transformations, `sample` will be introduced since it will be useful in many statistical learning scenarios. Then we will compare results with the `takeSample` action.      "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Getting the data and creating the RDD"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this case we will use the complete dataset provided for the KDD Cup 1999, containing nearly half million network interactions. The file is provided as a Gzip file that we will download locally."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import urllib\n",
    "f = urllib.urlretrieve (\"http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data.gz\", \"kddcup.data.gz\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can use this file to create our RDD."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "data_file = \"./kddcup.data.gz\"\n",
    "raw_data = sc.textFile(data_file)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sampling RDDs   "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In Spark, there are two sampling operations, the transformation `sample` and the action `takeSample`. By using a transformation we can tell Spark to apply successive transformation on a sample of a given RDD. By using an action we retrieve a given sample and we can have it in local memory to be used by any other standard library (e.g. Scikit-learn).  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The `sample` transformation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `sample` transformation takes up to three parameters. First is whether the sampling is done with replacement or not. Second is the sample size as a fraction. Finally we can optionally provide a *random seed*.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Sample size is 489957 of 4898431\n"
     ]
    }
   ],
   "source": [
    "raw_data_sample = raw_data.sample(False, 0.1, 1234)\n",
    "sample_size = raw_data_sample.count()\n",
    "total_size = raw_data.count()\n",
    "print \"Sample size is {} of {}\".format(sample_size, total_size)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "But the power of sampling as a transformation comes from doing it as part of a sequence of additional transformations. This will show more powerful once we start doing aggregations and key-value pairs operations, and will be specially useful when using Spark's machine learning library MLlib.    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the meantime, imagine we want to have an approximation of the proportion of `normal.` interactions in our dataset. We could do this by counting the total number of tags as we did in previous notebooks. However we want a quicker response and we don't need the exact answer but just an approximation. We can do it as follows.   "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The ratio of 'normal' interactions is 0.199\n",
      "Count done in 44.523 seconds\n"
     ]
    }
   ],
   "source": [
    "from time import time\n",
    "\n",
    "# transformations to be applied\n",
    "raw_data_sample_items = raw_data_sample.map(lambda x: x.split(\",\"))\n",
    "sample_normal_tags = raw_data_sample_items.filter(lambda x: \"normal.\" in x)\n",
    "\n",
    "# actions + time\n",
    "t0 = time()\n",
    "sample_normal_tags_count = sample_normal_tags.count()\n",
    "tt = time() - t0\n",
    "\n",
    "sample_normal_ratio = sample_normal_tags_count / float(sample_size)\n",
    "print \"The ratio of 'normal' interactions is {}\".format(round(sample_normal_ratio,3)) \n",
    "print \"Count done in {} seconds\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's compare this with calculating the ratio without sampling.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The ratio of 'normal' interactions is 0.199\n",
      "Count done in 91.09 seconds\n"
     ]
    }
   ],
   "source": [
    "# transformations to be applied\n",
    "raw_data_items = raw_data.map(lambda x: x.split(\",\"))\n",
    "normal_tags = raw_data_items.filter(lambda x: \"normal.\" in x)\n",
    "\n",
    "# actions + time\n",
    "t0 = time()\n",
    "normal_tags_count = normal_tags.count()\n",
    "tt = time() - t0\n",
    "\n",
    "normal_ratio = normal_tags_count / float(total_size)\n",
    "print \"The ratio of 'normal' interactions is {}\".format(round(normal_ratio,3)) \n",
    "print \"Count done in {} seconds\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can see a gain in time. The more transformations we apply after the sampling the bigger this gain. This is because without sampling all the transformations are applied to the complete set of data.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The `takeSample` action  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If what we need is to grab a sample of raw data from our RDD into local memory in order to be used by other non-Spark libraries, `takeSample` can be used.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The syntax is very similar, but in this case we specify the number of items instead of the sample size as a fraction of the complete data size.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The ratio of 'normal' interactions is 0.1988025\n",
      "Count done in 76.166 seconds\n"
     ]
    }
   ],
   "source": [
    "t0 = time()\n",
    "raw_data_sample = raw_data.takeSample(False, 400000, 1234)\n",
    "normal_data_sample = [x.split(\",\") for x in raw_data_sample if \"normal.\" in x]\n",
    "tt = time() - t0\n",
    "\n",
    "normal_sample_size = len(normal_data_sample)\n",
    "\n",
    "normal_ratio = normal_sample_size / 400000.0\n",
    "print \"The ratio of 'normal' interactions is {}\".format(normal_ratio)\n",
    "print \"Count done in {} seconds\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The process was very similar as before. We obtained a sample of about 10 percent of the data, and then filter and split.  \n",
    "\n",
    "However, it took longer, even with a slightly smaller sample. The reason is that Spark just distributed the execution of the sampling process. The filtering and splitting of the results were done locally in a single node.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Set operations on 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 supports many of the operations we have in mathematical sets, such as union and intersection, even when the RDDs themselves are not properly sets. It is important to note that these operations require that the RDDs being operated on are of the same type.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Set operations are quite straightforward to understand as it work as expected. The only consideration comes from the fact that RDDs are not real sets, and therefore operations such as the union of RDDs doesn't remove duplicates. In this notebook we will have a brief look at `subtract`, `distinct`, and `cartesian`.       "
   ]
  },
  {
   "cell_type": "markdown",
   "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, containing nearly half million network interactions. The file is provided as a Gzip file that we will download locally."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import urllib\n",
    "f = urllib.urlretrieve (\"http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz\", \"kddcup.data_10_percent.gz\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "data_file = \"./kddcup.data_10_percent.gz\"\n",
    "raw_data = sc.textFile(data_file)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Getting attack interactions using `subtract`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For illustrative purposes, imagine we already have our RDD with non attack (normal) interactions from some previous analysis.   "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "normal_raw_data = raw_data.filter(lambda x: \"normal.\" in x)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can obtain attack interactions by subtracting normal ones from the original unfiltered RDD as follows.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "attack_raw_data = raw_data.subtract(normal_raw_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's do some counts to check our results.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "All count in 5.261 secs\n"
     ]
    }
   ],
   "source": [
    "from time import time\n",
    "\n",
    "# count all\n",
    "t0 = time()\n",
    "raw_data_count = raw_data.count()\n",
    "tt = time() - t0\n",
    "print \"All count in {} secs\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Normal count in 5.571 secs\n"
     ]
    }
   ],
   "source": [
    "# count normal\n",
    "t0 = time()\n",
    "normal_raw_data_count = normal_raw_data.count()\n",
    "tt = time() - t0\n",
    "print \"Normal count in {} secs\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Attack count in 12.075 secs\n"
     ]
    }
   ],
   "source": [
    "# count attacks\n",
    "t0 = time()\n",
    "attack_raw_data_count = attack_raw_data.count()\n",
    "tt = time() - t0\n",
    "print \"Attack count in {} secs\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "There are 97278 normal interactions and 396743 attacks, from a total of 494021 interactions\n"
     ]
    }
   ],
   "source": [
    "print \"There are {} normal interactions and {} attacks, \\\n",
    "from a total of {} interactions\".format(normal_raw_data_count,attack_raw_data_count,raw_data_count)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So now we have two RDDs, one with normal interactions and another one with attacks.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Protocol and service combinations using `cartesian`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can compute the Cartesian product between two RDDs by using the `cartesian` transformation. It returns all possible pairs of elements between two RDDs. In our case we will use it to generate all the possible combinations between service and protocol in our network interactions.  \n",
    "\n",
    "First of all we need to isolate each collection of values in two separate RDDs. For that we will use `distinct` on the CSV-parsed dataset. From the [dataset description](http://kdd.ics.uci.edu/databases/kddcup99/kddcup.names) we know that protocol is the second column and service is the third (tag is the last one and not the first as appears in the page).   "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So first, let's get the protocols.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'udp', u'icmp', u'tcp']"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "csv_data = raw_data.map(lambda x: x.split(\",\"))\n",
    "protocols = csv_data.map(lambda x: x[1]).distinct()\n",
    "protocols.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we do the same for services.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'domain',\n",
       " u'http_443',\n",
       " u'Z39_50',\n",
       " u'smtp',\n",
       " u'urp_i',\n",
       " u'private',\n",
       " u'echo',\n",
       " u'shell',\n",
       " u'red_i',\n",
       " u'eco_i',\n",
       " u'sunrpc',\n",
       " u'ftp_data',\n",
       " u'urh_i',\n",
       " u'pm_dump',\n",
       " u'pop_3',\n",
       " u'pop_2',\n",
       " u'systat',\n",
       " u'ftp',\n",
       " u'uucp',\n",
       " u'whois',\n",
       " u'netbios_dgm',\n",
       " u'efs',\n",
       " u'remote_job',\n",
       " u'daytime',\n",
       " u'ntp_u',\n",
       " u'finger',\n",
       " u'ldap',\n",
       " u'netbios_ns',\n",
       " u'kshell',\n",
       " u'iso_tsap',\n",
       " u'ecr_i',\n",
       " u'nntp',\n",
       " u'printer',\n",
       " u'domain_u',\n",
       " u'uucp_path',\n",
       " u'courier',\n",
       " u'exec',\n",
       " u'time',\n",
       " u'netstat',\n",
       " u'telnet',\n",
       " u'gopher',\n",
       " u'rje',\n",
       " u'sql_net',\n",
       " u'link',\n",
       " u'auth',\n",
       " u'netbios_ssn',\n",
       " u'csnet_ns',\n",
       " u'X11',\n",
       " u'IRC',\n",
       " u'tftp_u',\n",
       " u'login',\n",
       " u'supdup',\n",
       " u'name',\n",
       " u'nnsp',\n",
       " u'mtp',\n",
       " u'http',\n",
       " u'bgp',\n",
       " u'ctf',\n",
       " u'hostnames',\n",
       " u'klogin',\n",
       " u'vmnet',\n",
       " u'tim_i',\n",
       " u'discard',\n",
       " u'imap4',\n",
       " u'other',\n",
       " u'ssh']"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "services = csv_data.map(lambda x: x[2]).distinct()\n",
    "services.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A longer list in this case."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can do the cartesian product.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "There are 198 combinations of protocol X service\n"
     ]
    }
   ],
   "source": [
    "product = protocols.cartesian(services).collect()\n",
    "print \"There are {} combinations of protocol X service\".format(len(product))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Obviously, for such small RDDs doesn't really make sense to use Spark cartesian product. We could have perfectly collected the values after using `distinct` and do the cartesian product locally. Moreover, `distinct` and `cartesian` are expensive operations so they must be used with care when the operating datasets are large.    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data aggregations on 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": [
    "We can aggregate RDD data in Spark by using three different actions: `reduce`, `fold`, and `aggregate`. The last one is the more general one and someway includes the first two.  "
   ]
  },
  {
   "cell_type": "markdown",
   "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 nework interactions. The file is provided as a Gzip file that we will download locally.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import urllib\n",
    "f = urllib.urlretrieve (\"http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz\", \"kddcup.data_10_percent.gz\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "data_file = \"./kddcup.data_10_percent.gz\"\n",
    "raw_data = sc.textFile(data_file)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Inspecting interaction duration by tag"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Both `fold` and `reduce` take a function as an argument that is applied to two elements of the RDD. The `fold` action differs from `reduce` in that it gets and additional initial *zero value* to be used for the initial call. This value should be the identity element for the function provided.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As an example, imagine we want to know the total duration of our interactions for normal and attack interactions. We can use `reduce` as follows.    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# parse data\n",
    "csv_data = raw_data.map(lambda x: x.split(\",\"))\n",
    "\n",
    "# separate into different RDDs\n",
    "normal_csv_data = csv_data.filter(lambda x: x[41]==\"normal.\")\n",
    "attack_csv_data = csv_data.filter(lambda x: x[41]!=\"normal.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The function that we pass to `reduce` gets and returns elements of the same type of the RDD. If we want to sum durations we need to extract that element into a new RDD.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "normal_duration_data = normal_csv_data.map(lambda x: int(x[0]))\n",
    "attack_duration_data = attack_csv_data.map(lambda x: int(x[0]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can reduce these new RDDs.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total duration for 'normal' interactions is 21075991\n",
      "Total duration for 'attack' interactions is 2626792\n"
     ]
    }
   ],
   "source": [
    "total_normal_duration = normal_duration_data.reduce(lambda x, y: x + y)\n",
    "total_attack_duration = attack_duration_data.reduce(lambda x, y: x + y)\n",
    "\n",
    "print \"Total duration for 'normal' interactions is {}\".\\\n",
    "    format(total_normal_duration)\n",
    "print \"Total duration for 'attack' interactions is {}\".\\\n",
    "    format(total_attack_duration)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can go further and use counts to calculate duration means.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Mean duration for 'normal' interactions is 216.657\n",
      "Mean duration for 'attack' interactions is 6.621\n"
     ]
    }
   ],
   "source": [
    "normal_count = normal_duration_data.count()\n",
    "attack_count = attack_duration_data.count()\n",
    "\n",
    "print \"Mean duration for 'normal' interactions is {}\".\\\n",
    "    format(round(total_normal_duration/float(normal_count),3))\n",
    "print \"Mean duration for 'attack' interactions is {}\".\\\n",
    "    format(round(total_attack_duration/float(attack_count),3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We have a first (and too simplistic) approach to identify attack interactions."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## A better way, using `aggregate`  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `aggregate` action frees us from the constraint of having the return be the same type as the RDD we are working on. Like with `fold`, we supply an initial zero value of the type we want to return. Then we provide two functions. The first one is used to combine the elements from our RDD with the accumulator. The second function is needed to merge two accumulators. Let's see it in action calculating the mean we did before.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Mean duration for 'normal' interactions is 216.657\n"
     ]
    }
   ],
   "source": [
    "normal_sum_count = normal_duration_data.aggregate(\n",
    "    (0,0), # the initial value\n",
    "    (lambda acc, value: (acc[0] + value, acc[1] + 1)), # combine value with acc\n",
    "    (lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])) # combine accumulators\n",
    ")\n",
    "\n",
    "print \"Mean duration for 'normal' interactions is {}\".\\\n",
    "    format(round(normal_sum_count[0]/float(normal_sum_count[1]),3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the previous aggregation, the accumulator first element keeps the total sum, while the second element keeps the count. Combining an accumulator with an RDD element consists in summing up the value and incrementing the count. Combining two accumulators requires just a pairwise sum.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can do the same with attack type interactions.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Mean duration for 'attack' interactions is 6.621\n"
     ]
    }
   ],
   "source": [
    "attack_sum_count = attack_duration_data.aggregate(\n",
    "    (0,0), # the initial value\n",
    "    (lambda acc, value: (acc[0] + value, acc[1] + 1)), # combine value with acc\n",
    "    (lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])) # combine accumulators\n",
    ")\n",
    "\n",
    "print \"Mean duration for 'attack' interactions is {}\".\\\n",
    "    format(round(attack_sum_count[0]/float(attack_sum_count[1]),3))"
   ]
  },
  {
   "cell_type": "markdown",
   "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": "markdown",
   "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",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import urllib\n",
    "f = urllib.urlretrieve (\"http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz\", \"kddcup.data_10_percent.gz\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "data_file = \"./kddcup.data_10_percent.gz\"\n",
    "raw_data = sc.textFile(data_file)"
   ]
  },
  {
   "cell_type": "markdown",
   "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",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "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"
   ]
  },
  {
   "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",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(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.'])]"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "key_value_data.take(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "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",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(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)]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "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()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We have a specific counting action for key/value pairs.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "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})"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "counts_by_key = key_value_data.countByKey()\n",
    "counts_by_key"
   ]
  },
  {
   "cell_type": "markdown",
   "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",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{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)}"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "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()"
   ]
  },
  {
   "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",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "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"
     ]
    }
   ],
   "source": [
    "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]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A small step into understanding what makes a network interaction be considered an attack."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spark SQL and Data Frames"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[Introduction to Spark with Python, by Jose A. Dianes](http://jadianes.github.io/spark-py-notebooks)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook will introduce Spark capabilities to deal with data in a structured way. Basically, everything turns around the concept of *Data Frame* and using *SQL language* to query them. We will see how the data frame abstraction, very popular in other data analytics ecosystems (e.g. R and Python/Pandas), it is very powerful when performing exploratory data analysis. In fact, it is very easy to express data queries when used together with the SQL language. Moreover, Spark distributes this column-based data structure transparently, in order to make the querying process as efficient as possible.      "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Getting the data and creating the RDD"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we did in previous notebooks, 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 nework interactions. The file is provided as a Gzip file that we will download locally.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import urllib\n",
    "f = urllib.urlretrieve (\"http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz\", \"kddcup.data_10_percent.gz\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "data_file = \"./kddcup.data_10_percent.gz\"\n",
    "raw_data = sc.textFile(data_file).cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Getting a Data Frame"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A Spark `DataFrame` is a distributed collection of data organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R or Pandas. They can be constructed from a wide array of sources such as a existing RDD in our case."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The entry point into all SQL functionality in Spark is the `SQLContext` class. To create a basic instance, all we need is a `SparkContext` reference. Since we are running Spark in shell mode (using pySpark) we can use the global context object `sc` for this purpose.    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Inferring the schema"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "With a `SQLContext`, we are ready to create a `DataFrame` from our existing RDD. But first we need to tell Spark SQL the schema in our data.   "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark SQL can convert an RDD of `Row` objects to a `DataFrame`. Rows are constructed by passing a list of key/value pairs as *kwargs* to the `Row` class. The keys define the column names, and the types are inferred by looking at the first row. Therefore, it is important that there is no missing data in the first row of the RDD in order to properly infer the schema."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In our case, we first need to split the comma separated data, and then use the information in KDD's 1999 task description to obtain the [column names](http://kdd.ics.uci.edu/databases/kddcup99/kddcup.names).  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import Row\n",
    "\n",
    "csv_data = raw_data.map(lambda l: l.split(\",\"))\n",
    "row_data = csv_data.map(lambda p: Row(\n",
    "    duration=int(p[0]), \n",
    "    protocol_type=p[1],\n",
    "    service=p[2],\n",
    "    flag=p[3],\n",
    "    src_bytes=int(p[4]),\n",
    "    dst_bytes=int(p[5])\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once we have our RDD of `Row` we can infer and register the schema.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "interactions_df = sqlContext.createDataFrame(row_data)\n",
    "interactions_df.registerTempTable(\"interactions\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can run SQL queries over our data frame that has been registered as a table.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "duration dst_bytes\n",
      "5057     0        \n",
      "5059     0        \n",
      "5051     0        \n",
      "5056     0        \n",
      "5051     0        \n",
      "5039     0        \n",
      "5062     0        \n",
      "5041     0        \n",
      "5056     0        \n",
      "5064     0        \n",
      "5043     0        \n",
      "5061     0        \n",
      "5049     0        \n",
      "5061     0        \n",
      "5048     0        \n",
      "5047     0        \n",
      "5044     0        \n",
      "5063     0        \n",
      "5068     0        \n",
      "5062     0        \n"
     ]
    }
   ],
   "source": [
    "# Select tcp network interactions with more than 1 second duration and no transfer from destination\n",
    "tcp_interactions = sqlContext.sql(\"\"\"\n",
    "    SELECT duration, dst_bytes FROM interactions WHERE protocol_type = 'tcp' AND duration > 1000 AND dst_bytes = 0\n",
    "\"\"\")\n",
    "tcp_interactions.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The results of SQL queries are RDDs and support all the normal RDD operations.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Duration: 5057, Dest. bytes: 0\n",
      "Duration: 5059, Dest. bytes: 0\n",
      "Duration: 5051, Dest. bytes: 0\n",
      "Duration: 5056, Dest. bytes: 0\n",
      "Duration: 5051, Dest. bytes: 0\n",
      "Duration: 5039, Dest. bytes: 0\n",
      "Duration: 5062, Dest. bytes: 0\n",
      "Duration: 5041, Dest. bytes: 0\n",
      "Duration: 5056, Dest. bytes: 0\n",
      "Duration: 5064, Dest. bytes: 0\n",
      "Duration: 5043, Dest. bytes: 0\n",
      "Duration: 5061, Dest. bytes: 0\n",
      "Duration: 5049, Dest. bytes: 0\n",
      "Duration: 5061, Dest. bytes: 0\n",
      "Duration: 5048, Dest. bytes: 0\n",
      "Duration: 5047, Dest. bytes: 0\n",
      "Duration: 5044, Dest. bytes: 0\n",
      "Duration: 5063, Dest. bytes: 0\n",
      "Duration: 5068, Dest. bytes: 0\n",
      "Duration: 5062, Dest. bytes: 0\n",
      "Duration: 5046, Dest. bytes: 0\n",
      "Duration: 5052, Dest. bytes: 0\n",
      "Duration: 5044, Dest. bytes: 0\n",
      "Duration: 5054, Dest. bytes: 0\n",
      "Duration: 5039, Dest. bytes: 0\n",
      "Duration: 5058, Dest. bytes: 0\n",
      "Duration: 5051, Dest. bytes: 0\n",
      "Duration: 5032, Dest. bytes: 0\n",
      "Duration: 5063, Dest. bytes: 0\n",
      "Duration: 5040, Dest. bytes: 0\n",
      "Duration: 5051, Dest. bytes: 0\n",
      "Duration: 5066, Dest. bytes: 0\n",
      "Duration: 5044, Dest. bytes: 0\n",
      "Duration: 5051, Dest. bytes: 0\n",
      "Duration: 5036, Dest. bytes: 0\n",
      "Duration: 5055, Dest. bytes: 0\n",
      "Duration: 2426, Dest. bytes: 0\n",
      "Duration: 5047, Dest. bytes: 0\n",
      "Duration: 5057, Dest. bytes: 0\n",
      "Duration: 5037, Dest. bytes: 0\n",
      "Duration: 5057, Dest. bytes: 0\n",
      "Duration: 5062, Dest. bytes: 0\n",
      "Duration: 5051, Dest. bytes: 0\n",
      "Duration: 5051, Dest. bytes: 0\n",
      "Duration: 5053, Dest. bytes: 0\n",
      "Duration: 5064, Dest. bytes: 0\n",
      "Duration: 5044, Dest. bytes: 0\n",
      "Duration: 5051, Dest. bytes: 0\n",
      "Duration: 5033, Dest. bytes: 0\n",
      "Duration: 5066, Dest. bytes: 0\n",
      "Duration: 5063, Dest. bytes: 0\n",
      "Duration: 5056, Dest. bytes: 0\n",
      "Duration: 5042, Dest. bytes: 0\n",
      "Duration: 5063, Dest. bytes: 0\n",
      "Duration: 5060, Dest. bytes: 0\n",
      "Duration: 5056, Dest. bytes: 0\n",
      "Duration: 5049, Dest. bytes: 0\n",
      "Duration: 5043, Dest. bytes: 0\n",
      "Duration: 5039, Dest. bytes: 0\n",
      "Duration: 5041, Dest. bytes: 0\n",
      "Duration: 42448, Dest. bytes: 0\n",
      "Duration: 42088, Dest. bytes: 0\n",
      "Duration: 41065, Dest. bytes: 0\n",
      "Duration: 40929, Dest. bytes: 0\n",
      "Duration: 40806, Dest. bytes: 0\n",
      "Duration: 40682, Dest. bytes: 0\n",
      "Duration: 40571, Dest. bytes: 0\n",
      "Duration: 40448, Dest. bytes: 0\n",
      "Duration: 40339, Dest. bytes: 0\n",
      "Duration: 40232, Dest. bytes: 0\n",
      "Duration: 40121, Dest. bytes: 0\n",
      "Duration: 36783, Dest. bytes: 0\n",
      "Duration: 36674, Dest. bytes: 0\n",
      "Duration: 36570, Dest. bytes: 0\n",
      "Duration: 36467, Dest. bytes: 0\n",
      "Duration: 36323, Dest. bytes: 0\n",
      "Duration: 36204, Dest. bytes: 0\n",
      "Duration: 32038, Dest. bytes: 0\n",
      "Duration: 31925, Dest. bytes: 0\n",
      "Duration: 31809, Dest. bytes: 0\n",
      "Duration: 31709, Dest. bytes: 0\n",
      "Duration: 31601, Dest. bytes: 0\n",
      "Duration: 31501, Dest. bytes: 0\n",
      "Duration: 31401, Dest. bytes: 0\n",
      "Duration: 31301, Dest. bytes: 0\n",
      "Duration: 31194, Dest. bytes: 0\n",
      "Duration: 31061, Dest. bytes: 0\n",
      "Duration: 30935, Dest. bytes: 0\n",
      "Duration: 30835, Dest. bytes: 0\n",
      "Duration: 30735, Dest. bytes: 0\n",
      "Duration: 30619, Dest. bytes: 0\n",
      "Duration: 30518, Dest. bytes: 0\n",
      "Duration: 30418, Dest. bytes: 0\n",
      "Duration: 30317, Dest. bytes: 0\n",
      "Duration: 30217, Dest. bytes: 0\n",
      "Duration: 30077, Dest. bytes: 0\n",
      "Duration: 25420, Dest. bytes: 0\n",
      "Duration: 22921, Dest. bytes: 0\n",
      "Duration: 22821, Dest. bytes: 0\n",
      "Duration: 22721, Dest. bytes: 0\n",
      "Duration: 22616, Dest. bytes: 0\n",
      "Duration: 22516, Dest. bytes: 0\n",
      "Duration: 22416, Dest. bytes: 0\n",
      "Duration: 22316, Dest. bytes: 0\n",
      "Duration: 22216, Dest. bytes: 0\n",
      "Duration: 21987, Dest. bytes: 0\n",
      "Duration: 21887, Dest. bytes: 0\n",
      "Duration: 21767, Dest. bytes: 0\n",
      "Duration: 21661, Dest. bytes: 0\n",
      "Duration: 21561, Dest. bytes: 0\n",
      "Duration: 21455, Dest. bytes: 0\n",
      "Duration: 21334, Dest. bytes: 0\n",
      "Duration: 21223, Dest. bytes: 0\n",
      "Duration: 21123, Dest. bytes: 0\n",
      "Duration: 20983, Dest. bytes: 0\n",
      "Duration: 14682, Dest. bytes: 0\n",
      "Duration: 14420, Dest. bytes: 0\n",
      "Duration: 14319, Dest. bytes: 0\n",
      "Duration: 14198, Dest. bytes: 0\n",
      "Duration: 14098, Dest. bytes: 0\n",
      "Duration: 13998, Dest. bytes: 0\n",
      "Duration: 13898, Dest. bytes: 0\n",
      "Duration: 13796, Dest. bytes: 0\n",
      "Duration: 13678, Dest. bytes: 0\n",
      "Duration: 13578, Dest. bytes: 0\n",
      "Duration: 13448, Dest. bytes: 0\n",
      "Duration: 13348, Dest. bytes: 0\n",
      "Duration: 13241, Dest. bytes: 0\n",
      "Duration: 13141, Dest. bytes: 0\n",
      "Duration: 13033, Dest. bytes: 0\n",
      "Duration: 12933, Dest. bytes: 0\n",
      "Duration: 12833, Dest. bytes: 0\n",
      "Duration: 12733, Dest. bytes: 0\n",
      "Duration: 12001, Dest. bytes: 0\n",
      "Duration: 5678, Dest. bytes: 0\n",
      "Duration: 5010, Dest. bytes: 0\n",
      "Duration: 1298, Dest. bytes: 0\n",
      "Duration: 1031, Dest. bytes: 0\n",
      "Duration: 36438, Dest. bytes: 0\n"
     ]
    }
   ],
   "source": [
    "# Output duration together with dst_bytes\n",
    "tcp_interactions_out = tcp_interactions.map(lambda p: \"Duration: {}, Dest. bytes: {}\".format(p.duration, p.dst_bytes))\n",
    "for ti_out in tcp_interactions_out.collect():\n",
    "  print ti_out"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can easily have a look at our data frame schema using `printSchema`.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- dst_bytes: long (nullable = true)\n",
      " |-- duration: long (nullable = true)\n",
      " |-- flag: string (nullable = true)\n",
      " |-- protocol_type: string (nullable = true)\n",
      " |-- service: string (nullable = true)\n",
      " |-- src_bytes: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "interactions_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Queries as `DataFrame` operations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark `DataFrame` provides a domain-specific language for structured data manipulation. This language includes methods we can concatenate in order to do selection, filtering, grouping, etc. For example, let's say we want to count how many interactions are there for each protocol type. We can proceed as follows.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "protocol_type count \n",
      "udp           20354 \n",
      "tcp           190065\n",
      "icmp          283602\n",
      "Query performed in 20.568 seconds\n"
     ]
    }
   ],
   "source": [
    "from time import time\n",
    "\n",
    "t0 = time()\n",
    "interactions_df.select(\"protocol_type\", \"duration\", \"dst_bytes\").groupBy(\"protocol_type\").count().show()\n",
    "tt = time() - t0\n",
    "\n",
    "print \"Query performed in {} seconds\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now imagine that we want to count how many interactions last more than 1 second, with no data transfer from destination, grouped by protocol type. We can just add to filter calls to the previous.   "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "protocol_type count\n",
      "tcp           139  \n",
      "Query performed in 16.641 seconds\n"
     ]
    }
   ],
   "source": [
    "t0 = time()\n",
    "interactions_df.select(\"protocol_type\", \"duration\", \"dst_bytes\").filter(interactions_df.duration>1000).filter(interactions_df.dst_bytes==0).groupBy(\"protocol_type\").count().show()\n",
    "tt = time() - t0\n",
    "\n",
    "print \"Query performed in {} seconds\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can use this to perform some [exploratory data analysis](http://en.wikipedia.org/wiki/Exploratory_data_analysis). Let's count how many attack and normal interactions we have. First we need to add the label column to our data.    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def get_label_type(label):\n",
    "    if label!=\"normal.\":\n",
    "        return \"attack\"\n",
    "    else:\n",
    "        return \"normal\"\n",
    "    \n",
    "row_labeled_data = csv_data.map(lambda p: Row(\n",
    "    duration=int(p[0]), \n",
    "    protocol_type=p[1],\n",
    "    service=p[2],\n",
    "    flag=p[3],\n",
    "    src_bytes=int(p[4]),\n",
    "    dst_bytes=int(p[5]),\n",
    "    label=get_label_type(p[41])\n",
    "    )\n",
    ")\n",
    "interactions_labeled_df = sqlContext.createDataFrame(row_labeled_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This time we don't need to register the schema since we are going to use the OO query interface.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's check the previous actually works by counting attack and normal data in our data frame.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "label  count \n",
      "attack 396743\n",
      "normal 97278 \n",
      "Query performed in 17.325 seconds\n"
     ]
    }
   ],
   "source": [
    "t0 = time()\n",
    "interactions_labeled_df.select(\"label\").groupBy(\"label\").count().show()\n",
    "tt = time() - t0\n",
    "\n",
    "print \"Query performed in {} seconds\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we want to count them by label and protocol type, in order to see how important the protocol type is to detect when an interaction is or not an attack.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "label  protocol_type count \n",
      "attack udp           1177  \n",
      "attack tcp           113252\n",
      "attack icmp          282314\n",
      "normal udp           19177 \n",
      "normal tcp           76813 \n",
      "normal icmp          1288  \n",
      "Query performed in 17.253 seconds\n"
     ]
    }
   ],
   "source": [
    "t0 = time()\n",
    "interactions_labeled_df.select(\"label\", \"protocol_type\").groupBy(\"label\", \"protocol_type\").count().show()\n",
    "tt = time() - t0\n",
    "\n",
    "print \"Query performed in {} seconds\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At first sight it seems that *udp* interactions are in lower proportion between network attacks versus other protocol types.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And we can do much more sophisticated groupings. For example, add to the previous a \"split\" based on data transfer from target.   "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "label  protocol_type (dst_bytes = 0) count \n",
      "normal icmp          true            1288  \n",
      "attack udp           true            1166  \n",
      "attack udp           false           11    \n",
      "normal udp           true            3594  \n",
      "normal udp           false           15583 \n",
      "attack tcp           true            110583\n",
      "attack tcp           false           2669  \n",
      "normal tcp           true            9313  \n",
      "normal tcp           false           67500 \n",
      "attack icmp          true            282314\n",
      "Query performed in 17.284 seconds\n"
     ]
    }
   ],
   "source": [
    "t0 = time()\n",
    "interactions_labeled_df.select(\"label\", \"protocol_type\", \"dst_bytes\").groupBy(\"label\", \"protocol_type\", interactions_labeled_df.dst_bytes==0).count().show()\n",
    "tt = time() - t0\n",
    "\n",
    "print \"Query performed in {} seconds\".format(round(tt,3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We see how relevant is this new split to determine if a network interaction is an attack.  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will stop here, but we can see how powerful this type of queries are in order to explore our data. Actually we can replicate all the splits we saw in previous notebooks, when introducing classification trees, just by selecting, groping, and filtering our dataframe. For a more detailed (but less real-world) list of Spark's `DataFrame` operations and data sources, have a look at the official documentation [here](https://spark.apache.org/docs/latest/sql-programming-guide.html#dataframe-operations).    "
   ]
  }
 ],
 "metadata": {
  "name": "_merged"
 },
 "nbformat": 4,
 "nbformat_minor": 2
}