nb4-rdd-set/nb4-rdd-set.ipynb

{
 "metadata": {
  "name": "",
  "signature": "sha256:123c139134363a65ac4461e7d98848e74ede7989fa222d57f2ff95d79405e114"
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "heading",
     "level": 1,
     "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": "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, 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": 2
    },
    {
     "cell_type": "heading",
     "level": 2,
     "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",
     "collapsed": false,
     "input": [
      "normal_raw_data = raw_data.filter(lambda x: \"normal.\" in x)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 3
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "We can obtain attack interactions by subtracting normal ones from the original unfiltered RDD as follows.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "attack_raw_data = raw_data.subtract(normal_raw_data)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 4
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Let's do some counts to check our results.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "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))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "All count in 5.261 secs\n"
       ]
      }
     ],
     "prompt_number": 5
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# 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))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "Normal count in 5.571 secs\n"
       ]
      }
     ],
     "prompt_number": 6
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# 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))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "Attack count in 12.075 secs\n"
       ]
      }
     ],
     "prompt_number": 7
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "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)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "There are 97278 normal interactions and 396743 attacks, from a total of 494021 interactions\n"
       ]
      }
     ],
     "prompt_number": 8
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "So now we have two RDDs, one with normal interactions and another one with attacks.  "
     ]
    },
    {
     "cell_type": "heading",
     "level": 2,
     "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",
     "collapsed": false,
     "input": [
      "csv_data = raw_data.map(lambda x: x.split(\",\"))\n",
      "protocols = csv_data.map(lambda x: x[1]).distinct()\n",
      "protocols.collect()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 9,
       "text": [
        "[u'udp', u'icmp', u'tcp']"
       ]
      }
     ],
     "prompt_number": 9
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now we do the same for services.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "services = csv_data.map(lambda x: x[2]).distinct()\n",
      "services.collect()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 10,
       "text": [
        "[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']"
       ]
      }
     ],
     "prompt_number": 10
    },
    {
     "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",
     "collapsed": false,
     "input": [
      "product = protocols.cartesian(services).collect()\n",
      "print \"There are {} combinations of protocol X service\".format(len(product))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "There are 198 combinations of protocol X service\n"
       ]
      }
     ],
     "prompt_number": 11
    },
    {
     "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.    "
     ]
    }
   ],
   "metadata": {}
  }
 ]
}