2.working_with_rdd_2.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2. Spark & Python: Working with RDDs (Il)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- DataSets :  KDD Cup 1999 ( http://kdd.ics.uci.edu/databases/kddcup99/kddcup99.html )\n",
    "\n",
    "This is the data set used for The Third International Knowledge Discovery and Data Mining Tools Competition, which was held in conjunction with KDD-99 The Fifth International Conference on Knowledge Discovery and Data Mining. The competition task was to build a network intrusion detector, a predictive model capable of distinguishing between ``bad'' connections, called intrusions or attacks, and ``good'' normal connections. This database contains a standard set of data to be audited, which includes a wide variety of intrusions simulated in a military network environment."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## RDD Creation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### (1) Getting the data files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import urllib\n",
    "\n",
    "f = urllib.request.urlretrieve(\"http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz\", \n",
    "                       \"kddcup.data_10_percent.gz\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### (2) Creating a RDD from a file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "data_file = \"./kddcup.data_10_percent.gz\"\n",
    "raw_data = sc.textFile(data_file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": true
   },
   "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": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# 첫 번째 value만 추출 -> int로 변환\n",
    "\n",
    "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": [
    "##  Inspecting interaction duration by tag"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**- normal**  \n",
    "**- attack**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### (1) reduce"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# normal/attact 기간 합계\n",
    "# rdd : action\n",
    "\n",
    "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)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "21075991\n",
      "2626792\n"
     ]
    }
   ],
   "source": [
    "print(total_normal_duration)\n",
    "print(total_attack_duration)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 127,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1 5 11 30 1\n"
     ]
    }
   ],
   "source": [
    "print(sc.parallelize([1, 2, 3, 5]).reduce(lambda x, y: x),\n",
    "      sc.parallelize([1, 2, 3, 5]).reduce(lambda x, y: y),\n",
    "      sc.parallelize([1, 2, 3, 5]).reduce(lambda x, y: x + y),\n",
    "      sc.parallelize([1, 2, 3, 5]).reduce(lambda x, y: x * y),\n",
    "      sc.parallelize([1, 2, 3, 5]).reduce(lambda x, y: x - y))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- reduce 연산 순서  \n",
    "https://kdb.or.kr/info/info_04_view.html?field=&keyword=&type=techreport&page=9&dbnum=186753&mode=detail&type=techreport"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### (2) count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "normal_count = normal_duration_data.count()\n",
    "attack_count = attack_duration_data.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "97278\n",
      "396743\n"
     ]
    }
   ],
   "source": [
    "print(normal_count)\n",
    "print(attack_count)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### (3) aggregate"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "normal_sum_count = normal_duration_data.aggregate(\n",
    "    (0,0), # 초기 값\n",
    "    (lambda acc, value: (acc[0] + value, acc[1] + 1)), # total sum (x + y), count\n",
    "    (lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])) # accumulators 합\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 133,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# 초기값 이유? -> 항등원"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 128,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(21075991, 97278)"
      ]
     },
     "execution_count": 128,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "normal_sum_count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "attack_sum_count = attack_duration_data.aggregate(\n",
    "    (0,0), # 초기값\n",
    "    (lambda acc, value: (acc[0] + value, acc[1] + 1)), \n",
    "    (lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])) \n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(21075991, 97278)\n",
      "(2626792, 396743)\n"
     ]
    }
   ],
   "source": [
    "print(normal_sum_count)\n",
    "print(attack_sum_count)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 150,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 150,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "attack_duration_data.getNumPartitions()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Working with key/value pair RDDs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "key_value_data = csv_data.map(lambda x: (x[41], x)) # x[41] contains the network interaction tag"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('normal.',\n",
       "  ['0',\n",
       "   'tcp',\n",
       "   'http',\n",
       "   'SF',\n",
       "   '181',\n",
       "   '5450',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '1',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '8',\n",
       "   '8',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '1.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '9',\n",
       "   '9',\n",
       "   '1.00',\n",
       "   '0.00',\n",
       "   '0.11',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   'normal.'])]"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "key_value_data.take(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### (1) Data aggregations with key/value pair RDDs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('portsweep.', 1991911.0),\n",
       " ('neptune.', 0.0),\n",
       " ('satan.', 64.0),\n",
       " ('pod.', 0.0),\n",
       " ('multihop.', 1288.0),\n",
       " ('back.', 284.0),\n",
       " ('warezclient.', 627563.0),\n",
       " ('nmap.', 0.0),\n",
       " ('smurf.', 0.0),\n",
       " ('guess_passwd.', 144.0),\n",
       " ('ftp_write.', 259.0),\n",
       " ('imap.', 72.0),\n",
       " ('land.', 0.0),\n",
       " ('loadmodule.', 326.0),\n",
       " ('buffer_overflow.', 2751.0),\n",
       " ('perl.', 124.0),\n",
       " ('ipsweep.', 43.0),\n",
       " ('rootkit.', 1008.0),\n",
       " ('phf.', 18.0),\n",
       " ('teardrop.', 0.0),\n",
       " ('warezmaster.', 301.0),\n",
       " ('normal.', 21075991.0),\n",
       " ('spy.', 636.0)]"
      ]
     },
     "execution_count": 18,
     "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": [
    "- key값 기준으로 reduce 연산"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 151,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from datetime import datetime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 153,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0:00:04.544392\n"
     ]
    }
   ],
   "source": [
    "s = datetime.now()\n",
    "\n",
    "durations_by_key = key_value_duration.reduceByKey(lambda x, y: x + y)\n",
    "durations_by_key.collect()\n",
    "\n",
    "e = datetime.now()\n",
    "print(e - s)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 스파크는 클러스터의 사이즈에 맞는 적절한 파티션 개수를 찾는 방식으로 동작함.\n",
    "- 퍼포먼스 향상을 위해 병렬화 수준 지정 가능\n",
    "- ex. key_value_duration.reduceByKey(lambda x, y: x + y, 4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "defaultdict(int,\n",
       "            {'back.': 2203,\n",
       "             'buffer_overflow.': 30,\n",
       "             'ftp_write.': 8,\n",
       "             'guess_passwd.': 53,\n",
       "             'imap.': 12,\n",
       "             'ipsweep.': 1247,\n",
       "             'land.': 21,\n",
       "             'loadmodule.': 9,\n",
       "             'multihop.': 7,\n",
       "             'neptune.': 107201,\n",
       "             'nmap.': 231,\n",
       "             'normal.': 97278,\n",
       "             'perl.': 3,\n",
       "             'phf.': 4,\n",
       "             'pod.': 264,\n",
       "             'portsweep.': 1040,\n",
       "             'rootkit.': 10,\n",
       "             'satan.': 1589,\n",
       "             'smurf.': 280790,\n",
       "             'spy.': 2,\n",
       "             'teardrop.': 979,\n",
       "             'warezclient.': 1020,\n",
       "             'warezmaster.': 20})"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "counts_by_key = key_value_data.countByKey()\n",
    "counts_by_key"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- key값 기준으로 count 연산"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### (2) Using combineByKey"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 168,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "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": "code",
   "execution_count": 173,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('portsweep.', (1991911.0, 1040)),\n",
       " ('neptune.', (0.0, 107201)),\n",
       " ('satan.', (64.0, 1589)),\n",
       " ('pod.', (0.0, 264)),\n",
       " ('multihop.', (1288.0, 7)),\n",
       " ('back.', (284.0, 2203)),\n",
       " ('warezclient.', (627563.0, 1020)),\n",
       " ('nmap.', (0.0, 231)),\n",
       " ('smurf.', (0.0, 280790)),\n",
       " ('guess_passwd.', (144.0, 53)),\n",
       " ('ftp_write.', (259.0, 8)),\n",
       " ('imap.', (72.0, 12)),\n",
       " ('land.', (0.0, 21)),\n",
       " ('loadmodule.', (326.0, 9)),\n",
       " ('buffer_overflow.', (2751.0, 30)),\n",
       " ('perl.', (124.0, 3)),\n",
       " ('ipsweep.', (43.0, 1247)),\n",
       " ('rootkit.', (1008.0, 10)),\n",
       " ('phf.', (18.0, 4)),\n",
       " ('teardrop.', (0.0, 979)),\n",
       " ('warezmaster.', (301.0, 20)),\n",
       " ('normal.', (21075991.0, 97278)),\n",
       " ('spy.', (636.0, 2))]"
      ]
     },
     "execution_count": 173,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sum_counts.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python [conda root]",
   "language": "python",
   "name": "conda-root-py"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}