msan694_disco/MSAN694-Lec4-Notes.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Lecture Notes MSAN694"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://10.1.45.27:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.2.0</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>PySparkShell</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local[*] appName=PySparkShell>"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## Quiz 3: Covers Material from Week 2-3-4\n",
    "\n",
    "Data - create the basic RDD and common python. Harder than last time, because more materials have been covered. Practice with the examples provided. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## From Week 3:\n",
    "\n",
    "### CombineByKey\n",
    "\n",
    "Similiar to aggregate(), but also includes the kkey and the values.\n",
    "\n",
    "\n",
    "**createCombiner** - initial value\n",
    "\n",
    "**mergeValue** - if not initial, then aggregator Partition\n",
    "\n",
    "**mergeCombiner** - this is adding across partitions\n",
    "\n",
    "```python\n",
    "'I love tacos'\n",
    "'I love coffee'\n",
    "'I love coffee'\n",
    "```\n",
    "\n",
    "**Target Output** \n",
    "\n",
    "```python\n",
    "(len, (occurence, word))\n",
    "```\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "** Step by Step **\n",
    "\n",
    "```python\n",
    "\n",
    ".textFile(' ')\n",
    ".flatMap(lambda x ; x.split())\n",
    ".map(lambda x: (len(x), x)\n",
    "\n",
    "output -> partition 1: (1,I) (4,love) (5,tacos), (1,I) (4,love) (6,coffee)\n",
    "output -> partition 2: (1,I) (4, love) (6, coffee)\n",
    "\n",
    "# SAMPLE FUNCTIONS (WHATS HAPPENING IN THE BACKGROUND)\n",
    "# will nest the tuple in another key\n",
    "createCombiner: lambda x: (1,x)\n",
    "mergevalue : lambda x,y : x[0] + 1, x[1] +', '+ y\n",
    "\n",
    "(1,I) (4,love) (5,tacos), (1,I) (4,love) (6,coffee)\n",
    "\n",
    "```\n",
    "#### Partition 1:\n",
    "```python\n",
    "(5, (1, tacos))\n",
    "(1, (2, I, I)\n",
    "(4, (1+1, love, love))\n",
    "(6, (1, coffee))\n",
    "```\n",
    "#### Partition 2:\n",
    "```python\n",
    "(1, (1, I))\n",
    "(4, (1, love))\n",
    "(6, (1, coffee))\n",
    "```\n",
    "\n",
    "#### How do you combine between partitions? \n",
    "\n",
    "```python\n",
    "mergeCombiner: lambda x,y : x[0] + y[0], x[1] + \",\" + y[1]\n",
    "```\n",
    "\n",
    "#### All together\n",
    "\n",
    "Before the data comes in, its already in (length, word) format\n",
    "```python\n",
    "(2, hi)\n",
    "(4, love)\n",
    "...\n",
    "```\n",
    "\n",
    "Then run this function\n",
    "\n",
    "```python\n",
    "comp = words.combineByKey(\n",
    "                    (lambda x : (1,x)), \n",
    "                   (lambda x,y : (x[0] +1 , x[1] +', '+ y)), \n",
    "                   (lambda x,y : (x[0] + y[0], x[1] +', '+y[1])))\n",
    "```\n",
    "\n",
    "##### First it will nest the base value\n",
    "```python\n",
    "(key, value)\n",
    "lambda x : (1,x)\n",
    "(key, (occurance(default=1), growing list of words))\n",
    "\n",
    "```\n",
    "\n",
    "\n",
    "```python\n",
    "(2 (1, hi))\n",
    "```\n",
    "\n",
    "##### Let's add (2,to), it has the same key so it will combine:\n",
    "\n",
    "```python\n",
    "(2 (1, hi))\n",
    "lambda x,y : (x[0] +1 , x[1] +', '+ y)\n",
    "(2 (2, hi, to))\n",
    "```\n",
    "##### Let's add (3,bye), it has the same key so it will combine:\n",
    "\n",
    "```python\n",
    "(2 (2, hi, to))\n",
    "(3 (1, bye))\n",
    "```\n",
    "\n",
    "##### And so on\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Week 3: Example 7"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "I love tacos\n",
       "I love coffee\n",
       "I love coffee\n",
       " MapPartitionsRDD[5] at textFile at NativeMethodAccessorImpl.java:0"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lines = sc.textFile(text)\n",
    "lines"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "PATH = '/Users/tlee010/Desktop/github_repos/2017-msan694-example/Data/README.md'\n",
    "lines = sc.textFile(PATH)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "words = lines.flatMap(lambda x : x. split(' '))\n",
    "words = words.map(lambda x : (len(x),x))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1, u'#'),\n",
       " (6, u'Apache'),\n",
       " (5, u'Spark'),\n",
       " (0, u''),\n",
       " (5, u'Spark'),\n",
       " (2, u'is'),\n",
       " (1, u'a'),\n",
       " (4, u'fast'),\n",
       " (3, u'and')]"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "words.collect()[:9]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(0,\n",
       "  (68,\n",
       "   u', , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , , ')),\n",
       " (112,\n",
       "  (1,\n",
       "   u'[Eclipse](https://cwiki.apache.org/confluence/display/SPARK/Useful+Developer+Tools#UsefulDeveloperTools-Eclipse)')),\n",
       " (2,\n",
       "  (70,\n",
       "   u'is, It, in, R,, an, It, of, ##, on, ##, is, To, do, to, do, if, by, -T, in, is, at, an, ##, to, is, to, ##, if, ##, in, To, of, Pi, to, to, be, or, to, on, to, or, to, an, if, is, in, of, if, no, ##, is, be, on, to, or, ##, to, to, in, of, to, at, on, of, ##, to, in, an, on, to')),\n",
       " (4,\n",
       "  (47,\n",
       "   u'fast, APIs, that, data, also, rich, find, This, file, only, run:, (You, need, this, more, than, with, More, from, IDE,, also, also, with, will, when, This, URL,, with, with, also, name, Many, help, Once, [run, Note, uses, core, talk, HDFS, have, must, same, that, your, Hive, Hive')),\n",
       " (6,\n",
       "  (41,\n",
       "   u'Apache, system, Scala,, engine, graphs, GraphX, stream, Online, latest, guide,, README, thread, option, Maven,, builds, shell:, should, return, scala>, Python, prefer, Python, shell:, should, return, sample, MASTER, submit, \"yarn\", params, given., built,, using:, Please, Hadoop, Hadoop, Please, Hadoop, Please, online, Spark.')),\n",
       " (8,\n",
       "  (33,\n",
       "   u'provides, supports, supports, [project, [project, contains, Building, detailed, command,, command,, Programs, programs, example:, locally., variable, examples, examples, cluster., mesos://, spark://, threads., package., programs, requires, guidance, Versions, systems., versions, detailed, guidance, building, building, overview'))]"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "comp = words.combineByKey(\n",
    "                    (lambda x : (1,x)), \n",
    "                   (lambda x,y : (x[0] +1 , x[1] +', '+ y)), \n",
    "                   (lambda x,y : (x[0] + y[0], x[1] +', '+y[1])))\n",
    "\n",
    "comp.collect()[:6]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Question 8:\n",
    "\n",
    "Which operations don't  shuffle data?\n",
    "\n",
    "\n",
    "    - mapValues() <- O WITHIN the partition applies\n",
    "    - groupByKey() <-- X will shuffle ( up front)\n",
    "    - reduceByKey() <-- X will shuffle (at the end) \n",
    "    - combineBykey() <-- X will shuffle (at the end)\n",
    "    - sortByKey() <-- X will shuffle\n",
    "    - flatMapValues() <- O WITHIN the partition applies"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Week4 : Pair RDD\n",
    "\n",
    "- Pair RDD\n",
    "- Pair RDD Operations - Actions\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Actions \n",
    "- subtractByKey\n",
    "- join\n",
    "- rightOuterJoin\n",
    "- leftOuterJoin"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Subtract by Key\n",
    "- Only compares the keys\n",
    "\n",
    "#### Example:\n",
    "\n",
    "```python\n",
    "A = [(2,3), (1,2), (1,3)]\n",
    "B = [(1,2)]\n",
    "```\n",
    "\n",
    "#### Note that it ignores the actual values in the pairs, only compares the keys\n",
    "```python\n",
    "A.subtract(B) = [(2,3), (1,3)]\n",
    "A.subtractByKey(B) = [(2,3)]\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Week4 : Example 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "List business names wihtout a district supervisor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "PATH1 = '/Users/tlee010/Desktop/github_repos/2017-msan694-example/Data/supervisor_sf.csv'\n",
    "PATH2 = '/Users/tlee010/Desktop/github_repos/2017-msan694-example/Data/filtered_registered_business_sf.csv'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "lines1 = sc.textFile(PATH1)\n",
    "lines2 = sc.textFile(PATH2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'94102', u'8'),\n",
       " (u'94102', u'6'),\n",
       " (u'94102', u'3'),\n",
       " (u'94102', u'5'),\n",
       " (u'94103', u'8')]"
      ]
     },
     "execution_count": 86,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# split by word\n",
    "zippairs = lines1.map(lambda x : x.split(','))\n",
    "\n",
    "# then turn into tuple \n",
    "zippairs = zippairs.map(lambda x : (x[0],x[1]))\n",
    "\n",
    "# look at data\n",
    "zippairs.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'94123', u'Tournahu George L'),\n",
       " (u'94124', u'Stephens Institute Inc'),\n",
       " (u'94105', u'Stephens Institute Inc'),\n",
       " (u'94108', u'Stephens Institute Inc'),\n",
       " (u'94107', u'Stephens Institute Inc')]"
      ]
     },
     "execution_count": 87,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# split the 2nd file by commas\n",
    "zipbus = lines2.map(lambda x : x.split(','))\n",
    "\n",
    "# split the 2nd file by commas\n",
    "zipbus = zipbus.map(lambda x : (x[0], x[1]))\n",
    "zipbus.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'Precision Communication Serv',\n",
       " u'Schefer Thomas R',\n",
       " u'Tyrrell Thomas',\n",
       " u'Lucid Systems',\n",
       " u'Jacob Abraham']"
      ]
     },
     "execution_count": 88,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "res = zipbus.subtractByKey(zippairs).values().distinct()\n",
    "res.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Week 4 : Join , rightOuterjoin, LeftOuterjoin\n",
    "## Week 4 : cogroup(otherdataset)\n",
    "\n",
    "Group data from both RDDs sharing the same key. Go over two RDDs sharing the same key. Return the key and respective lists from two RDD values.\n",
    "\n",
    "##### Group by Illustration\n",
    "```python\n",
    ".groupByKey()\n",
    "\n",
    "a = [(1,2),(1,3)] -- > group by key --> [1, <iterable>] = [1, [2,3]]\n",
    "\n",
    "b = [(2,2),(2,3)] --> group by key -- > [2, <iterable>] = [2, [2,3]]\n",
    "```\n",
    "\n",
    "##### Cogroup sample\n",
    "```python\n",
    ".cogroup(1 other dataset)\n",
    "\n",
    "(1, (<iterable>,<iterable>)) = (1, ([2,3],[3]))\n",
    "\n",
    "(2, (<iterable>,<iterable>)) = (2, ([],[2,3])) \n",
    "\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Week4 : Example 2\n",
    "\n",
    "Try out the join commands"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "first_num_pairs = sc.parallelize({(2,3), (1,2),(1,3),(2,4),(3,6)})\n",
    "\n",
    "second_num_pairs = sc.parallelize({(1,3),(2,2)})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1, 2), (1, 3), (2, 3), (2, 4), (3, 6)]"
      ]
     },
     "execution_count": 94,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "first_num_pairs.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1, 3), (2, 2)]"
      ]
     },
     "execution_count": 95,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "second_num_pairs.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(2, 3), (3, 6), (1, 2), (2, 4)]"
      ]
     },
     "execution_count": 97,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "first_num_pairs.subtract(second_num_pairs).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(3, 6)]"
      ]
     },
     "execution_count": 98,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "first_num_pairs.subtractByKey(second_num_pairs).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1, (2, 3)), (1, (3, 3)), (2, (3, 2)), (2, (4, 2))]"
      ]
     },
     "execution_count": 99,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "first_num_pairs.join(second_num_pairs).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Note that its reversed, RIGHT = keep the originals, add additiona new data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1, (2, 3)), (1, (3, 3)), (2, (3, 2)), (2, (4, 2))]"
      ]
     },
     "execution_count": 100,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "first_num_pairs.rightOuterJoin(second_num_pairs).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Note that its reversed, LEFT = only return rows found in new dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1, (2, 3)), (1, (3, 3)), (2, (3, 2)), (2, (4, 2)), (3, (6, None))]"
      ]
     },
     "execution_count": 101,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "first_num_pairs.leftOuterJoin(second_num_pairs).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1,\n",
       "  (<pyspark.resultiterable.ResultIterable at 0x117073790>,\n",
       "   <pyspark.resultiterable.ResultIterable at 0x115c6bc50>)),\n",
       " (2,\n",
       "  (<pyspark.resultiterable.ResultIterable at 0x115c6b850>,\n",
       "   <pyspark.resultiterable.ResultIterable at 0x115c6b890>)),\n",
       " (3,\n",
       "  (<pyspark.resultiterable.ResultIterable at 0x115c6b310>,\n",
       "   <pyspark.resultiterable.ResultIterable at 0x115c6b050>))]"
      ]
     },
     "execution_count": 102,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "first_num_pairs.cogroup(second_num_pairs).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Week4 : Example 3\n",
    "\n",
    "Using Example 1 list (zip (business_name, supervisor_id) pairs ordered  y supervisor_id\n",
    "- only if both business and supervisor exist\n",
    "- if a business exists\n",
    "- if a supervisor exists"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'94102,8', u'94102,6', u'94102,3', u'94102,5', u'94103,8']"
      ]
     },
     "execution_count": 103,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lines1.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 104,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'94123,Tournahu George L,3301 Broderick St,San Francisco,CA',\n",
       " u'94124,Stephens Institute Inc,2225 Jerrold Ave,San Francisco,CA',\n",
       " u'94105,Stephens Institute Inc,180 New Montgomery St,San Francisco,CA',\n",
       " u'94108,Stephens Institute Inc,540 Powell St,San Francisco,CA',\n",
       " u'94107,Stephens Institute Inc,460 Townsend St,San Francisco,CA']"
      ]
     },
     "execution_count": 104,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lines2.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'94123', u'Tournahu George L'),\n",
       " (u'94124', u'Stephens Institute Inc'),\n",
       " (u'94105', u'Stephens Institute Inc'),\n",
       " (u'94108', u'Stephens Institute Inc')]"
      ]
     },
     "execution_count": 106,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "zipbus = lines2.map(lambda x : x.split(','))\n",
    "zipbus = zipbus.map(lambda x : (x[0],x[1]))\n",
    "zipbus.take(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'94102', u'8'), (u'94102', u'6'), (u'94102', u'3'), (u'94102', u'5')]"
      ]
     },
     "execution_count": 110,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "zips_super = lines1.map(lambda x : x.split(','))\n",
    "zips_super = zips_super.map(lambda x : (x[0],x[1]))\n",
    "zips_super.take(4)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### If both business and supervisor exist (inner join)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 123,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'94117', (u'Alouis Auto Radiator Inc', u'1')),\n",
       " (u'94117', (u'Atlantic Richfield Co', u'1')),\n",
       " (u'94117', (u'Breall William Etal', u'1')),\n",
       " (u'94117', (u'Cole Hardware Inc', u'1')),\n",
       " (u'94117', (u'Cole Hardware Inc', u'1'))]"
      ]
     },
     "execution_count": 123,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "zip_comb = zipbus.join(zips_super).sortBy(lambda x : x[1][1])\n",
    "zip_comb.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### If a business exists"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 126,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'94117', (u'Alouis Auto Radiator Inc', u'1')),\n",
       " (u'94117', (u'Atlantic Richfield Co', u'1')),\n",
       " (u'94117', (u'Breall William Etal', u'1')),\n",
       " (u'94117', (u'Cole Hardware Inc', u'1')),\n",
       " (u'94117', (u'Cole Hardware Inc', u'1'))]"
      ]
     },
     "execution_count": 126,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "zip_comb = zipbus.rightOuterJoin(zips_super).sortBy(lambda x : x[1][1])\n",
    "zip_comb.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### If a supervisor exists"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 127,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'', (u'Hartmann Studios Incorporated', None)),\n",
       " (u'', (u'Cardno Entrix', None)),\n",
       " (u'', (u'Bond Blacktop Inc', None)),\n",
       " (u'', (u'Moonka Nishi', None)),\n",
       " (u'', (u'Cooper Jim B', None))]"
      ]
     },
     "execution_count": 127,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "zip_comb = zipbus.leftOuterJoin(zips_super).sortBy(lambda x : x[1][1])\n",
    "zip_comb.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Week4 : Example 4\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Genearte Key(zip) and value pair RDDs from “filtered_registered_business_sf.csv” and “supervisor_sf.csv” and cogroup() the RDDs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 130,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'94102', u'8'),\n",
       " (u'94102', u'6'),\n",
       " (u'94102', u'3'),\n",
       " (u'94102', u'5'),\n",
       " (u'94103', u'8')]"
      ]
     },
     "execution_count": 130,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "superzips = lines1.map(lambda x: x.split(','))\n",
    "superzips = superzips.map(lambda x : (x[0],x[1]))\n",
    "superzips.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 131,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'94123', u'Tournahu George L'),\n",
       " (u'94124', u'Stephens Institute Inc'),\n",
       " (u'94105', u'Stephens Institute Inc'),\n",
       " (u'94108', u'Stephens Institute Inc'),\n",
       " (u'94107', u'Stephens Institute Inc')]"
      ]
     },
     "execution_count": 131,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "businesszips = lines2.map(lambda x: x.split(','))\n",
    "businesszips = businesszips.map(lambda x : (x[0],x[1]))\n",
    "businesszips.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 134,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'',\n",
       "  (<pyspark.resultiterable.ResultIterable at 0x1163a5ed0>,\n",
       "   <pyspark.resultiterable.ResultIterable at 0x1163ab0d0>)),\n",
       " (u'85716',\n",
       "  (<pyspark.resultiterable.ResultIterable at 0x1163ab310>,\n",
       "   <pyspark.resultiterable.ResultIterable at 0x1163ab350>)),\n",
       " (u'28036',\n",
       "  (<pyspark.resultiterable.ResultIterable at 0x1163ab390>,\n",
       "   <pyspark.resultiterable.ResultIterable at 0x1163ab3d0>)),\n",
       " (u'97302',\n",
       "  (<pyspark.resultiterable.ResultIterable at 0x1163ab410>,\n",
       "   <pyspark.resultiterable.ResultIterable at 0x1163ab450>)),\n",
       " (u'55344',\n",
       "  (<pyspark.resultiterable.ResultIterable at 0x1163ab510>,\n",
       "   <pyspark.resultiterable.ResultIterable at 0x1163ab550>)),\n",
       " (u'94035',\n",
       "  (<pyspark.resultiterable.ResultIterable at 0x1163ab590>,\n",
       "   <pyspark.resultiterable.ResultIterable at 0x1163ab5d0>))]"
      ]
     },
     "execution_count": 134,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "superzips.cogroup(businesszips).take(6)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 136,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "defaultdict(int,\n",
       "            {u'94102': 4,\n",
       "             u'94103': 6,\n",
       "             u'94104': 2,\n",
       "             u'94105': 2,\n",
       "             u'94107': 2,\n",
       "             u'94108': 2,\n",
       "             u'94109': 4,\n",
       "             u'94110': 4,\n",
       "             u'94111': 2,\n",
       "             u'94112': 5,\n",
       "             u'94114': 3,\n",
       "             u'94115': 3,\n",
       "             u'94116': 2,\n",
       "             u'94117': 4,\n",
       "             u'94118': 3,\n",
       "             u'94121': 2,\n",
       "             u'94122': 4,\n",
       "             u'94123': 1,\n",
       "             u'94124': 2,\n",
       "             u'94127': 3,\n",
       "             u'94129': 1,\n",
       "             u'94130': 1,\n",
       "             u'94131': 3,\n",
       "             u'94132': 3,\n",
       "             u'94133': 2,\n",
       "             u'94134': 3,\n",
       "             u'94158': 2})"
      ]
     },
     "execution_count": 136,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "superzips.countByKey()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 137,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'7', u'8', u'11', u'9', u'10']"
      ]
     },
     "execution_count": 137,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "superzips.lookup('94112')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}