w261/wk7/MIDS-W261-2015-HWK-Week07-CarinLlopSatpati-groupZ.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "#DATASCI W261: Machine Learning at Scale"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#HW7 (Group Z)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Juanjo Carin, Christopher Llop, Sayantan Satpati\n",
    "- [juanjose.carin@ischool.berkeley.edu](mailto:juanjose.carin@ischol.berkeley.com)\n",
    "- [christopher.llop@ischool.berkeley.edu](mailto:christopher.llop@ischool.berkeley.edu)\n",
    "- [sayantan.satpati@ischool.berkeley.edu](mailto:sayantan.satpati@ischool.berkeley.edu)\n",
    "- W261-2\n",
    "- Week 07\n",
    "- Submission date: 10/27/2015"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#General Description\n",
    "\n",
    "**In this assignment you will explore networks and develop MRJob code for finding shortest path graph distances. To build up to large data you will develop your code on some very simple, toy networks. After this you will take your developed code forward and modify it and apply it to two larger datasets (performing EDA along the way).**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#Undirected toy network dataset\n",
    "\n",
    "**In an undirected network all links are symmetric, i.e., for a pair of nodes 'A' and 'B,' both of the links A -> B and B -> A will exist.**\n",
    "\n",
    "**The toy data are available in a sparse (stripes) representation:**\n",
    "\n",
    "    (node) \\t (dictionary of links)\n",
    "\n",
    "**on AWS via the url: s3://ucb-mids-mls-networks/undirected_toy.txt**\n",
    "\n",
    "**In the dictionary, target nodes are keys, link weights are values (here, all weights are 1, i.e., the network is unweighted).**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#Directed toy network dataset\n",
    "\n",
    "**In a directed network all links are not necessarily symmetric, i.e., for a pair of nodes 'A' and 'B,' it is possible for only one of A -> B or B -> A to exist.**\n",
    "\n",
    "**These toy data are available in a sparse (stripes) representation:**\n",
    "\n",
    "    (node) \\t (dictionary of links)\n",
    "\n",
    "**on AWS via the url: s3://ucb-mids-mls-networks/directed_toy.txt**\n",
    "\n",
    "**In the dictionary, target nodes are keys, link weights are values (here, all weights are 1, i.e., the network is unweighted).**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#HW7.0: Shortest path graph distances (toy networks)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**In this part of your assignment you will develop the base of your code for the week.**\n",
    "\n",
    "**Write MRJob classes to find shortest path graph distances, as described in the lectures. In addition to finding the distances, your code should also output a distance-minimizing path between the source and target. Work locally for this part of the assignment, and use both of the undirected and directed toy networks.**\n",
    "\n",
    "**To proof you code's function, run the following jobs**\n",
    "\n",
    "- **shortest path in the undirected network from node 1 to node 4\n",
    "    - Solution: 1,5,4 \n",
    "- shortest path in the directed network from node 1 to node 5\n",
    "    - Solution: 1,2,4,5**\n",
    "\n",
    "**and report your output---make sure it is correct!**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from collections import defaultdict\n",
    "class Graph:\n",
    "    def __init__(self):\n",
    "        self.nodes = set()\n",
    "        self.edges = defaultdict(list)\n",
    "        self.distances = {}\n",
    " \n",
    "    def add_node(self, value):\n",
    "        self.nodes.add(value)\n",
    " \n",
    "    def add_edge(self, node1, node2, distance = 1,direct = True):\n",
    "        self.edges[node1].append(node2)\n",
    "        self.distances[(node1, node2)] = distance\n",
    "        if not direct:\n",
    "            self.edges[node2].append(node1)\n",
    "            self.distances[(node2, node1)] = distance\n",
    " \n",
    " \n",
    "def dijsktra(graph, initial):\n",
    "    visited = {initial: 0}\n",
    "    nodes = set(graph.nodes)\n",
    "    while nodes:\n",
    "        min_node = None\n",
    "        for node in nodes:\n",
    "            if node in visited:\n",
    "                if min_node is None:\n",
    "                    min_node = node\n",
    "                elif visited[node] < visited[min_node]:\n",
    "                    min_node = node\n",
    "        if min_node is None:\n",
    "            break\n",
    "        nodes.remove(min_node)\n",
    "        current_weight = visited[min_node]\n",
    "        for neighbour in graph.edges[min_node]:\n",
    "            try:\n",
    "                weight = current_weight + graph.distances[(min_node, neighbour)]\n",
    "            except:\n",
    "                continue\n",
    "            if neighbour not in visited or weight < visited[neighbour]:\n",
    "                visited[neighbour] = weight\n",
    "    return visited"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['1', '2', '3', '4', '5']\n",
      "[('1', '2', 1), ('1', '6', 1), ('2', '1', 1), ('2', '3', 1), ('2', '4', 1), ('3', '2', 1), ('3', '4', 1), ('4', '2', 1), ('4', '5', 1), ('5', '1', 1), ('5', '2', 1), ('5', '4', 1)]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "{'1': 0, '2': 1, '3': 2, '4': 2, '5': 3, '6': 1}"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import ast\n",
    "g = Graph()\n",
    "nodes = []\n",
    "edges = []\n",
    "with open('directed_toy.txt', 'r') as myfile:\n",
    "    for line in myfile:\n",
    "        word = line.split('\\t')\n",
    "        nodes.append(word[0])\n",
    "        dict = ast.literal_eval(word[1])\n",
    "        for k in dict.keys():\n",
    "            edges.append((word[0],k,dict[k]))  \n",
    "print nodes\n",
    "print edges\n",
    "#nodes = ['A', 'B', 'C', 'D', 'E', 'F']\n",
    "#edges = [('A', 'B', 1), ('A', 'C', 5), ('B', 'C', 2), ('C', 'D', 4), ('C', 'E', 3), ('D', 'F', 5), ('F', 'C', 3)]\n",
    "for node in nodes:\n",
    "    g.add_node(node)\n",
    "for edge in edges:\n",
    "    g.add_edge(*edge)\n",
    "dijsktra(g, '1')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['1']\n",
      "{'1': 0, '3': inf, '2': inf, '5': inf, '4': inf}\n",
      "2 1\n",
      "6 1\n"
     ]
    }
   ],
   "source": [
    "source = '1'\n",
    "SSSP = {}\n",
    "for node in nodes:\n",
    "    if node == source:\n",
    "        SSSP[node] = 0\n",
    "    else:\n",
    "        SSSP[node] = float('inf')\n",
    "Frontiers = [source]\n",
    "print Frontiers\n",
    "print SSSP\n",
    "\n",
    "import ast\n",
    "emul_reducer = {}\n",
    "with open('directed_toy.txt', 'r') as myfile:\n",
    "    for line in myfile:\n",
    "        line = line.split('\\t')\n",
    "        node = line[0]\n",
    "        sink = ast.literal_eval(line[1])\n",
    "        for sink_node in sink.keys():\n",
    "            if node in Frontiers:\n",
    "                #yield sink_node, SSSP[node] + sink[sink_node]\n",
    "                emul_reducer[sink_node] = SSSP[node] + sink[sink_node]\n",
    "\n",
    "for k in emul_reducer.keys():\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#HW5.1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "1. **In the database world What is 3NF? **\n",
    "2. **Does machine learning use data in 3NF? If so why?**\n",
    "3. **In what form does ML consume data?**\n",
    "4. **Why would one use log files that are denormalized?**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "1. **3NF** (Third Normal Form) is a normalization (i.e., an organization of data into columns--attributes--and tables--relations-- to minimize redundancies, by decomposing a flat table into smaller relational tables)  in which:\n",
    "    * the relation table is in 2NF:\n",
    "        * every non-prime attribute of the table (i.e., that does not belong to any candidate key of the table) is dependent on the whole of every candidate key)\n",
    "    * every non-prime attribute  of the relation table is non-transitively dependent on every superkey of R.\n",
    "\n",
    "  Requiring existence of \"the key\" ensures that the table is in 1NF; requiring that non-key attributes be dependent on \"the whole key\" ensures 2NF; further requiring that non-key attributes be dependent on \"nothing but the key\" ensures 3NF.\n",
    "\n",
    "2. To solve **Machine Learning** problems we usually do not use data in **3NF** because the information in each table alone does not give the \"full picture:\" we need to **denormalize** the **data** first, joining or aggregating tables, to be able to answer typical questions from a Machine Learning perspective (that involve all the dimensions at hand)\n",
    "\n",
    "3. As mentioned in the previous point, ML algorithms use **denormalized data**. This is because most of those algorithms apply a function repeatedly to the same dataset to optimize a parameter (e.g., through gradient descen), so we need the total amount of information.\n",
    "\n",
    "4. For the reason exposed above: denormalized log files include, for a particular observation (or log file), all the information (variables) we are going to use to apply a ML algorithm."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#HW5.2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Using MRJob, implement a hashside join (memory-backed map-side) for left, right and inner joins. Run your code on the  data used in HW 4.4: (Recall HW 4.4: Find the most frequent visitor of each page using mrjob and the output of 4.2 (i.e., transformed log file). In this output please include the webpage URL, webpageID and Visitor ID.)**\n",
    "\n",
    "**Justify which table you chose as the Left table in this hashside join.**\n",
    "\n",
    "**Please report the number of rows resulting from:**\n",
    "\n",
    "1. **Inner joining Table Left with Table Right**\n",
    "\n",
    "2. **Right joining Table Left with Table Right**\n",
    "\n",
    "3. **Left joining Table Left with Table Right**\n",
    "\n",
    "(I've reversed the order mentioned in the Instructions, so each new join adds a bit of complexity over the previous)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create Left and Right Tables\n",
    "Since I included the URLs in the transformed log file, I will generate both tables from scratch.\n",
    "\n",
    "Recall that the lines in the original file have these form:\n",
    "\n",
    "    ...\n",
    "    A,1100,1,\"MS in Education\",\"/education\"\n",
    "    A,1210,1,\"SNA Support\",\"/snasupport\"\n",
    "    C,\"10001\",10001\n",
    "    V,1000,1\n",
    "    V,1001,1\n",
    "    V,1002,1\n",
    "    C,\"10002\",10002\n",
    "    V,1001,1\n",
    "    V,1003,1\n",
    "    ...\n",
    "\n",
    "I.e., all the webpage IDs (the primary key) are listed with their URLs, and then each visitor ID, followed by the webpages he or she visited."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Training Instances  32711\n",
      "Attributes  294\n"
     ]
    }
   ],
   "source": [
    "import urllib2\n",
    "url = 'http://archive.ics.uci.edu/ml/machine-learning-databases/anonymous/' +\\\n",
    "    'anonymous-msweb.data'\n",
    "import os\n",
    "os.chdir('/home/hduser/Dropbox/W261/HW5')\n",
    "# Two counters to keep track of number of distinct webpages and visitors\n",
    "A = 0\n",
    "C = 0\n",
    "\n",
    "with open('TableLeft.txt', 'w') as TL, open('TableRight.txt', 'w') as TR:\n",
    "    for line in urllib2.urlopen(url):\n",
    "        record = line.strip().split(',')\n",
    "        record = [x.strip('\"') for x in record]\n",
    "        # If the record corresponds to an attribute, linke webpage ID with URL\n",
    "        if record[0] == 'A':\n",
    "            A += 1\n",
    "            key = record[1] # webpage ID\n",
    "            value = record[4] # webpage URL\n",
    "            TL.write(key + ',' + value + '\\n')\n",
    "        # If the record corresponds to a case (visitor), save that info...\n",
    "        elif record[0] == 'C':\n",
    "            C += 1\n",
    "            value = record[1]\n",
    "        # ... and pass it to the Vroot (i.e., link visitor ID and webpage ID)\n",
    "        elif record[0] == 'V':\n",
    "            key = record[1]\n",
    "            TR.write(key + ',' + value + '\\n')\n",
    "            \n",
    "print 'Training Instances  {}'.format(C)\n",
    "print 'Attributes  {}'.format(A)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "According to [https://kdd.ics.uci.edu/databases/msweb/msweb.data.html](https://kdd.ics.uci.edu/databases/msweb/msweb.data.html) there were:\n",
    "\n",
    "`Training Instances  32711\n",
    "Attributes  294`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Exploratory analysis of the 2 tables:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of webpages:         294\n",
      "Number of visitors:         32711\n",
      "Number of webpages visited: 285\n",
      "Number of visits:           98654\n"
     ]
    }
   ],
   "source": [
    "# Number of lines in TableLeft.txt\n",
    "!echo \"Number of webpages:         \"$(cat TableLeft.txt | wc -l)\n",
    "# Number of unique visitor IDs in TableRight.txt\n",
    "!echo \"Number of visitors:         \"$(cat TableRight.txt | cut -d',' -f2 | \\\n",
    "                                      uniq | wc -l)\n",
    "# Number of unique webpage IDs in TableRight.txt\n",
    "    # (sort before finding unique values: they have to be adjacent)\n",
    "!echo \"Number of webpages visited: \"$(cat TableRight.txt | cut -d',' -f1 | \\\n",
    "                                      sort | uniq | wc -l)\n",
    "# Number of lines in TableRight.txt\n",
    "!echo \"Number of visits:           \"$(cat TableRight.txt | wc -l)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we already saw in HW4, 9 webpages were not visited."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##HW5.2.1: Inner"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create MRJob task for Inner Joins"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting HashSideInnerJoin.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile HashSideInnerJoin.py\n",
    "from mrjob.job import MRJob\n",
    "import csv\n",
    "    \n",
    "class HashSideInnerJoin(MRJob):\n",
    "\n",
    "    def mapper_init(self):\n",
    "        # Load left-side table in memory as dictionary\n",
    "        self.TL = {}\n",
    "        # The absolute path will be passed as argument when calling MRJob\n",
    "        for key, value in csv.reader(open(\"TableLeft.txt\", \"r\")):\n",
    "            # key = webpage ID, value = webpage URL\n",
    "            self.TL[key] = value   \n",
    "        \n",
    "    def mapper(self, _, line):\n",
    "        # Iterate over the right-side table, a record at a time\n",
    "        TRrecord = line.split(\",\")\n",
    "        key = TRrecord[0]\n",
    "        value_visitor = TRrecord[1]\n",
    "        # Look for each record, in the left-side table (in-memory)\n",
    "        if key in self.TL.keys():\n",
    "            yield key, (self.TL[key], value_visitor)\n",
    "    \n",
    "    # The reducer is optional. If not specified, I found out records are not \n",
    "        # sorted by webpage ID\n",
    "    def reducer(self, key, value):\n",
    "        for val_url, val_visitor in value:\n",
    "            yield key, (val_url, val_visitor)\n",
    "            \n",
    "if __name__ == '__main__':\n",
    "    HashSideInnerJoin.run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create Python script to execute any of the 3 types of Join"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting HW52.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile HW52.py\n",
    "#!/home/hduser/anaconda/bin/python\n",
    "import sys\n",
    "JoinType = sys.argv[1]\n",
    "\n",
    "# Import the class\n",
    "\n",
    "if JoinType == 'Inner':\n",
    "    from HashSideInnerJoin import HashSideInnerJoin\n",
    "    JoinClass = 'HashSideInnerJoin'\n",
    "    output = 'InnerJoinTable.txt'\n",
    "elif JoinType == 'Right':\n",
    "    from HashSideRightJoin import HashSideRightJoin\n",
    "    JoinClass = 'HashSideRightJoin'\n",
    "    output = 'RightJoinTable.txt'\n",
    "elif JoinType == 'Left':\n",
    "    from HashSideLeftJoin import HashSideLeftJoin\n",
    "    JoinClass = 'HashSideLeftJoin'\n",
    "    output = 'LeftJoinTable.txt'\n",
    "else:\n",
    "    raise ValueError('USE Inner, Right, OR Left AS ARGUMENTS')\n",
    "    \n",
    "# Use the 2 tables, left-side as seconrd argument (to be load by mapper_init)\n",
    "mr_job = eval(JoinClass)(args=['TableRight.txt', '--file=TableLeft.txt'])\n",
    "with mr_job.make_runner() as runner:\n",
    "    runner.run()\n",
    "    # Create the join table\n",
    "    with open(output,'w') as result:\n",
    "        for line in runner.stream_output():\n",
    "            webpageID = str(mr_job.parse_output_line(line)[0])\n",
    "            # Extract webpage URL and visitor ID from value\n",
    "            webpageURL = mr_job.parse_output_line(line)[1][0]\n",
    "            visitorID = str(mr_job.parse_output_line(line)[1][1])\n",
    "            result.writelines(webpageID + ',' + webpageURL + ',' + visitorID \n",
    "                              +'\\n')\n",
    "    result.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###Call Python Script with Inner Join"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!chmod a+x HW52.py\n",
    "!./HW52.py Inner"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###Create bash script for EDA of the joint table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting EDA_HW52.sh\n"
     ]
    }
   ],
   "source": [
    "%%writefile EDA_HW52.sh\n",
    "joinTable=$1\n",
    "\n",
    "echo \"Number of webpage IDs:                            \"\\\n",
    "    $(cut -d, -f1 $joinTable | grep -v None | sort | uniq | wc -l)\n",
    "echo \"Number of webpage URLs:                           \"\\\n",
    "    $(cut -d, -f1,2 $joinTable | grep -v None | sort  | uniq | wc -l)\n",
    "echo \"Number of webpages with no associated webpage URL:\"\\\n",
    "    $(cut -d, -f1,2 $joinTable | grep None | sort | uniq | wc -l)\n",
    "echo \"Number of webpages visited:                       \"\\\n",
    "    $(cut -d, -f1,3 $joinTable | grep -v None | cut -d, -f1 | sort | uniq | \\\n",
    "      wc -l)\n",
    "echo \"Number of records:                                \"\\\n",
    "    $(wc -l < $joinTable)\n",
    "echo \"Number of visits:                                 \"\\\n",
    "    $(cut -d, -f3 $joinTable | grep -v None | wc -l)\n",
    "echo \"Number of webpages with no associated visitor ID: \"\\\n",
    "    $(cut -d, -f1,3 $joinTable | grep None | sort | uniq | wc -l)\n",
    "echo \"Number of visitors (IDs):                         \"\\\n",
    "    $(cut -d, -f3 $joinTable | grep -v None | sort | uniq | wc -l)\n",
    "if [ $(grep None $joinTable | wc -l) != 0 ]; then \\\n",
    "    echo -e \"Webpages with no visits or URL:\\n$(grep None $joinTable | \\\n",
    "    sort | sed 's/^/\\t/')\"; fi"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###EDA and output of the inner join"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Exploratory analysis of the joint table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of webpage IDs:                             285\n",
      "Number of webpage URLs:                            285\n",
      "Number of webpages with no associated webpage URL: 0\n",
      "Number of webpages visited:                        285\n",
      "Number of records:                                 98654\n",
      "Number of visits:                                  98654\n",
      "Number of webpages with no associated visitor ID:  0\n",
      "Number of visitors (IDs):                          32711\n"
     ]
    }
   ],
   "source": [
    "!chmod a+x EDA_HW52.sh\n",
    "!./EDA_HW52.sh InnerJoinTable.txt"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Of course (being this an inner join), all webpage URLs are matched with visitor IDs and vice versa."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1000,/regwiz,10001\r\n",
      "1000,/regwiz,10010\r\n",
      "1000,/regwiz,10039\r\n",
      "1000,/regwiz,10073\r\n",
      "1000,/regwiz,10087\r\n",
      "1000,/regwiz,10101\r\n",
      "1000,/regwiz,10132\r\n",
      "1000,/regwiz,10141\r\n",
      "1000,/regwiz,10154\r\n",
      "1000,/regwiz,10162\r\n",
      "1000,/regwiz,10166\r\n",
      "1000,/regwiz,10201\r\n",
      "1000,/regwiz,10218\r\n",
      "1000,/regwiz,10220\r\n",
      "1000,/regwiz,10324\r\n",
      "1000,/regwiz,10348\r\n",
      "1000,/regwiz,10376\r\n",
      "1000,/regwiz,10384\r\n",
      "1000,/regwiz,10409\r\n",
      "1000,/regwiz,10429\r\n",
      "1000,/regwiz,10454\r\n",
      "1000,/regwiz,10457\r\n",
      "1000,/regwiz,10471\r\n",
      "1000,/regwiz,10497\r\n",
      "1000,/regwiz,10511\r\n",
      "1000,/regwiz,10520\r\n",
      "1000,/regwiz,10541\r\n",
      "1000,/regwiz,10564\r\n",
      "1000,/regwiz,10599\r\n",
      "1000,/regwiz,10752\r\n",
      "1000,/regwiz,10756\r\n",
      "1000,/regwiz,10861\r\n",
      "1000,/regwiz,10935\r\n",
      "1000,/regwiz,10943\r\n",
      "1000,/regwiz,10969\r\n",
      "1000,/regwiz,11027\r\n",
      "1000,/regwiz,11050\r\n",
      "1000,/regwiz,11410\r\n",
      "1000,/regwiz,11429\r\n",
      "1000,/regwiz,11440\r\n",
      "1000,/regwiz,11490\r\n",
      "1000,/regwiz,11501\r\n",
      "1000,/regwiz,11528\r\n",
      "1000,/regwiz,11539\r\n",
      "1000,/regwiz,11544\r\n",
      "1000,/regwiz,11685\r\n",
      "1000,/regwiz,11695\r\n",
      "1000,/regwiz,11723\r\n",
      "1000,/regwiz,11766\r\n",
      "1000,/regwiz,11774\r\n"
     ]
    }
   ],
   "source": [
    "!head -50 InnerJoinTable.txt"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##HW5.2.2: Right"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create MRJob task for Right Joins"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting HashSideRightJoin.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile HashSideRightJoin.py\n",
    "from mrjob.job import MRJob\n",
    "import csv\n",
    "    \n",
    "class HashSideRightJoin(MRJob):\n",
    "\n",
    "    def mapper_init(self):\n",
    "        # Load left-side table in memory as dictionary\n",
    "        self.TL = {}\n",
    "        # The absolute path will be passed as argument when calling MRJob\n",
    "        for key, value in csv.reader(open(\"TableLeft.txt\", \"r\")):\n",
    "            # key = webpage ID, value = webpage URL\n",
    "            self.TL[key] = value   \n",
    "        \n",
    "    def mapper(self, _, line):\n",
    "        # Iterate over the right-side table, a record at a time\n",
    "        TRrecord = line.split(\",\")\n",
    "        key = TRrecord[0]\n",
    "        value_visitor = TRrecord[1]\n",
    "        # Look for each record, in the left-side table (in-memory)\n",
    "        if key in self.TL.keys():\n",
    "            yield key, (self.TL[key], value_visitor)\n",
    "        # And if there's no match, include the visitor info anyway\n",
    "        else:\n",
    "            yield key, (None, value_visitor)\n",
    "    \n",
    "    # The reducer is optional. If not specified, I found out records are not \n",
    "        # sorted by webpage ID\n",
    "    def reducer(self, key, value):\n",
    "        for val_url, val_visitor in value:\n",
    "            yield key, (val_url, val_visitor)\n",
    "            \n",
    "if __name__ == '__main__':\n",
    "    HashSideRightJoin.run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###Call Python Script with Right Join"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!python HW52.py Right"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###EDA and output of the right join"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Exploratory analysis of the joint table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of webpage IDs:                             285\n",
      "Number of webpage URLs:                            285\n",
      "Number of webpages visited:                        285\n",
      "Number of webpages with no associated webpage URL: 0\n",
      "Number of records:                                 98654\n",
      "Number of visits:                                  98654\n",
      "Number of webpages with no associated visitor ID:  0\n",
      "Number of visitors (IDs):                          32711\n"
     ]
    }
   ],
   "source": [
    "!./EDA_HW52.sh RightJoinTable.txt"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Being this a right join, we might have found some visits not matched with any URL, but that's not the case (because all primary keys: the webpage IDs) appear in the left-side table."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1000,/regwiz,10001\r\n",
      "1000,/regwiz,10010\r\n",
      "1000,/regwiz,10039\r\n",
      "1000,/regwiz,10073\r\n",
      "1000,/regwiz,10087\r\n",
      "1000,/regwiz,10101\r\n",
      "1000,/regwiz,10132\r\n",
      "1000,/regwiz,10141\r\n",
      "1000,/regwiz,10154\r\n",
      "1000,/regwiz,10162\r\n",
      "1000,/regwiz,10166\r\n",
      "1000,/regwiz,10201\r\n",
      "1000,/regwiz,10218\r\n",
      "1000,/regwiz,10220\r\n",
      "1000,/regwiz,10324\r\n",
      "1000,/regwiz,10348\r\n",
      "1000,/regwiz,10376\r\n",
      "1000,/regwiz,10384\r\n",
      "1000,/regwiz,10409\r\n",
      "1000,/regwiz,10429\r\n",
      "1000,/regwiz,10454\r\n",
      "1000,/regwiz,10457\r\n",
      "1000,/regwiz,10471\r\n",
      "1000,/regwiz,10497\r\n",
      "1000,/regwiz,10511\r\n",
      "1000,/regwiz,10520\r\n",
      "1000,/regwiz,10541\r\n",
      "1000,/regwiz,10564\r\n",
      "1000,/regwiz,10599\r\n",
      "1000,/regwiz,10752\r\n",
      "1000,/regwiz,10756\r\n",
      "1000,/regwiz,10861\r\n",
      "1000,/regwiz,10935\r\n",
      "1000,/regwiz,10943\r\n",
      "1000,/regwiz,10969\r\n",
      "1000,/regwiz,11027\r\n",
      "1000,/regwiz,11050\r\n",
      "1000,/regwiz,11410\r\n",
      "1000,/regwiz,11429\r\n",
      "1000,/regwiz,11440\r\n",
      "1000,/regwiz,11490\r\n",
      "1000,/regwiz,11501\r\n",
      "1000,/regwiz,11528\r\n",
      "1000,/regwiz,11539\r\n",
      "1000,/regwiz,11544\r\n",
      "1000,/regwiz,11685\r\n",
      "1000,/regwiz,11695\r\n",
      "1000,/regwiz,11723\r\n",
      "1000,/regwiz,11766\r\n",
      "1000,/regwiz,11774\r\n"
     ]
    }
   ],
   "source": [
    "!head -50 RightJoinTable.txt "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##HW5.2.3: Left"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create MRJob task for Left Joins"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting HashSideLeftJoin.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile HashSideLeftJoin.py\n",
    "from mrjob.job import MRJob\n",
    "import csv\n",
    "    \n",
    "class HashSideLeftJoin(MRJob):\n",
    "\n",
    "    def __init__(self, *args, **kwargs):\n",
    "        super(HashSideLeftJoin, self).__init__(*args, **kwargs)\n",
    "        self.TLkeys = []\n",
    "\n",
    "    def mapper_init(self):\n",
    "        # Load left-side table in memory as dictionary\n",
    "        self.TL = {}\n",
    "        # The absolute path will be passed as argument when calling MRJob\n",
    "        for key, value in csv.reader(open(\"TableLeft.txt\", \"r\")):\n",
    "            # key = webpage ID, value = webpage URL\n",
    "            self.TL[key] = value   \n",
    "            self.TLkeys.append(key)\n",
    "        \n",
    "    def mapper(self, _, line):\n",
    "        # Iterate over the right-side table, a record at a time\n",
    "        TRrecord = line.split(\",\")\n",
    "        key = TRrecord[0]\n",
    "        value_visitor = TRrecord[1]\n",
    "        # Look for each record, in the left-side table (in-memory)\n",
    "        if key in self.TL.keys():\n",
    "            try:\n",
    "                self.TLkeys.remove(key)\n",
    "            except ValueError:\n",
    "                pass\n",
    "            yield key, (self.TL[key], value_visitor)\n",
    "    \n",
    "    def mapper_final(self):\n",
    "        # Iterate over the right-side table, a record at a time\n",
    "        for key in self.TLkeys:\n",
    "            yield key, (self.TL[key], None)\n",
    "    \n",
    "    def reducer(self, key, value):\n",
    "        for val_url, val_visitor in value:\n",
    "            yield key, (val_url, val_visitor)\n",
    "            \n",
    "if __name__ == '__main__':\n",
    "    HashSideLeftJoin.run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###Call Python Script with Left Join"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!python HW52.py Left"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###EDA and output of the left join"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Exploratory analysis of the joint table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of webpage IDs:                             294\n",
      "Number of webpage URLs:                            294\n",
      "Number of webpages visited:                        285\n",
      "Number of webpages with no associated webpage URL: 0\n",
      "Number of records:                                 98663\n",
      "Number of visits:                                  98654\n",
      "Number of webpages with no associated visitor ID:  9\n",
      "Number of visitors (IDs):                          32711\n",
      "Webpages with no visits or URL:\n",
      "\t1287,/autoroute,None\n",
      "\t1288,/library,None\n",
      "\t1289,/masterchef,None\n",
      "\t1290,/devmovies,None\n",
      "\t1291,/news,None\n",
      "\t1292,/northafrica,None\n",
      "\t1293,/encarta,None\n",
      "\t1294,/bookshelf,None\n",
      "\t1297,/centroam,None\n"
     ]
    }
   ],
   "source": [
    "!./EDA_HW52.sh LeftJoinTable.txt"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As expected, this joint table contains 9 more records than the other two, since 9 URLs are not matched with any visitor IDs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1000,/regwiz,10001\r\n",
      "1000,/regwiz,10010\r\n",
      "1000,/regwiz,10039\r\n",
      "1000,/regwiz,10073\r\n",
      "1000,/regwiz,10087\r\n",
      "1000,/regwiz,10101\r\n",
      "1000,/regwiz,10132\r\n",
      "1000,/regwiz,10141\r\n",
      "1000,/regwiz,10154\r\n",
      "1000,/regwiz,10162\r\n",
      "1000,/regwiz,10166\r\n",
      "1000,/regwiz,10201\r\n",
      "1000,/regwiz,10218\r\n",
      "1000,/regwiz,10220\r\n",
      "1000,/regwiz,10324\r\n",
      "1000,/regwiz,10348\r\n",
      "1000,/regwiz,10376\r\n",
      "1000,/regwiz,10384\r\n",
      "1000,/regwiz,10409\r\n",
      "1000,/regwiz,10429\r\n",
      "1000,/regwiz,10454\r\n",
      "1000,/regwiz,10457\r\n",
      "1000,/regwiz,10471\r\n",
      "1000,/regwiz,10497\r\n",
      "1000,/regwiz,10511\r\n",
      "1000,/regwiz,10520\r\n",
      "1000,/regwiz,10541\r\n",
      "1000,/regwiz,10564\r\n",
      "1000,/regwiz,10599\r\n",
      "1000,/regwiz,10752\r\n",
      "1000,/regwiz,10756\r\n",
      "1000,/regwiz,10861\r\n",
      "1000,/regwiz,10935\r\n",
      "1000,/regwiz,10943\r\n",
      "1000,/regwiz,10969\r\n",
      "1000,/regwiz,11027\r\n",
      "1000,/regwiz,11050\r\n",
      "1000,/regwiz,11410\r\n",
      "1000,/regwiz,11429\r\n",
      "1000,/regwiz,11440\r\n",
      "1000,/regwiz,11490\r\n",
      "1000,/regwiz,11501\r\n",
      "1000,/regwiz,11528\r\n",
      "1000,/regwiz,11539\r\n",
      "1000,/regwiz,11544\r\n",
      "1000,/regwiz,11685\r\n",
      "1000,/regwiz,11695\r\n",
      "1000,/regwiz,11723\r\n",
      "1000,/regwiz,11766\r\n",
      "1000,/regwiz,11774\r\n"
     ]
    }
   ],
   "source": [
    "!head -50 LeftJoinTable.txt"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#HW5.3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## See the other notebooks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "155\tROPLEZIMPREDASTRODONBRASLPKLSON YHROACLMPARCHEYXMMIOUDAVESAURUS PIOFPILOCOWERSURUASOGETSESNEGCP TYRAVOPSIFENGOQUAPIALLOBOSKENUO OWINFUYAIOKENECKSASXHYILPOYNUAT\r\n",
      "155\tAIOPJUMRXUYVASLYHYPSIBEMAPODIKR UFRYDIUUOLBIGASUAURUSREXLISNAYE RNOONDQSRUNSUBUNOUGRABBERYAIRTC UTAHRAPTOREDILEIPMILBDUMMYUVERI SYEVRAHVELOCYALLOSAURUSLINROTSR\r\n"
     ]
    }
   ],
   "source": [
    "!python Longest_driver.py\n",
    "# No need to print first values: there is only one"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "5375699242\tthe\r\n",
      "3691308874\tof\r\n",
      "2221164346\tto\r\n",
      "1387638591\tin\r\n",
      "1342195425\ta\r\n",
      "1135779433\tand\r\n",
      "798553959\tthat\r\n",
      "756296656\tis\r\n",
      "688053106\tbe\r\n",
      "481373389\tas\r\n"
     ]
    }
   ],
   "source": [
    "!./FreqUnigrams_driver.py | sort -rn # Sort the 10 items in reverse order"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "11.557291666666666\t\"xxxx\"\r\n",
      "10.161726044782885\t\"NA\"\r\n",
      "8.0741599073001158\t\"blah\"\r\n",
      "7.5333333333333332\t\"nnn\"\r\n",
      "6.5611436445056839\t\"nd\"\r\n",
      "5.4073642846747196\t\"ND\"\r\n",
      "4.921875\t\"oooooooooooooooo\"\r\n",
      "4.7272727272727275\t\"PIC\"\r\n",
      "4.5116279069767442\t\"llll\"\r\n",
      "4.3494983277591972\t\"LUTHER\"\r\n",
      "4.2072378595731514\t\"oooooo\"\r\n",
      "4.0908402725208175\t\"NN\"\r\n",
      "3.9492846924177396\t\"ooooo\"\r\n",
      "3.9313725490196076\t\"OOOOOO\"\r\n",
      "3.7877030162412995\t\"IIII\"\r\n",
      "3.7624521072796937\t\"lillelu\"\r\n",
      "3.6570701447431206\t\"OOOOO\"\r\n",
      "3.6065624999999999\t\"Sc\"\r\n",
      "3.5769230769230771\t\"Pfeffermann\"\r\n",
      "3.5769230769230771\t\"Madarassy\"\r\n",
      "3.5600000000000001\t\"Meteoritical\"\r\n",
      "3.5364916773367479\t\"Undecided\"\r\n",
      "3.505639097744361\t\"Lib\"\r\n",
      "3.5\t\"xxxxxxxx\"\r\n",
      "3.4791318864774623\t\"ri\"\r\n",
      "3.3750684931506849\t\"Vir\"\r\n",
      "3.2390171258376768\t\"DREAM\"\r\n",
      "3.2290388548057258\t\"beep\"\r\n",
      "3.1886792452830188\t\"Latha\"\r\n",
      "3.1883175058233291\t\"MARTIN\"\r\n",
      "3.1699346405228757\t\"Lis\"\r\n",
      "3.1147458480120784\t\"Ac\"\r\n",
      "3.0371428571428569\t\"OUTPUT\"\r\n",
      "3.0222222222222221\t\"HENNESSY\"\r\n",
      "3.0\t\"ALLIS\"\r\n",
      "2.9191176470588234\t\"IYENGAR\"\r\n",
      "2.8698912704670052\t\"ft\"\r\n",
      "2.8432451923076925\t\"Adapted\"\r\n",
      "2.8250000000000002\t\"counterfeiteth\"\r\n",
      "2.8198198198198199\t\"nonsquamous\"\r\n",
      "2.8198198198198199\t\"nonmorular\"\r\n",
      "2.8085106382978724\t\"RHYME\"\r\n",
      "2.7446808510638299\t\"YOUTHS\"\r\n",
      "2.7264957264957266\t\"Poing\"\r\n",
      "2.7000000000000002\t\"Kuhl\"\r\n",
      "2.6748466257668713\t\"Sirignano\"\r\n",
      "2.6734693877551021\t\"YARDS\"\r\n",
      "2.6734693877551021\t\"METRES\"\r\n",
      "2.66414686825054\t\"Illl\"\r\n",
      "2.6603773584905661\t\"Neophytos\"\r\n",
      "sort: write failed: standard output: Broken pipe\r\n",
      "sort: write error\r\n"
     ]
    }
   ],
   "source": [
    "!sort -rn DenseUnigrams.txt | head -50"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of unigrams that never appear more than once in a page: 166114\n",
      "\n",
      "1.0\t\"Aana\"\n",
      "1.0\t\"AAN\"\n",
      "1.0\t\"Aan\"\n",
      "1.0\t\"aame\"\n",
      "1.0\t\"AAMC\"\n",
      "1.0\t\"Aaltonen\"\n",
      "1.0\t\"AAL\"\n",
      "1.0\t\"aahs\"\n",
      "1.0\t\"AAHPERD\"\n",
      "1.0\t\"aahed\"\n",
      "1.0\t\"aah\"\n",
      "1.0\t\"Aagje\"\n",
      "1.0\t\"AAFES\"\n",
      "1.0\t\"AAE\"\n",
      "1.0\t\"Aadam\"\n",
      "1.0\t\"AACVPR\"\n",
      "1.0\t\"AACP\"\n",
      "1.0\t\"AAAE\"\n",
      "1.0\t\"AAAA\"\n",
      "1.0\t\"aA\"\n"
     ]
    }
   ],
   "source": [
    "!echo -e \"Number of unigrams that never appear more than once in a page: \"\\\n",
    "    $(grep $'1.0\\t' DenseUnigrams.txt | wc -l)\"\\n\"\n",
    "!sort -rn DenseUnigrams.txt | tail -20"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#HW5.4"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**In this part of the assignment we will focus on developing methods for detecting synonyms, using the Google 5-grams dataset. To accomplish this you must script two main tasks using MRJob:**\n",
    "\n",
    "1. **Build stripes of word co-ocurrence for the top 10,000 most frequently appearing words across the entire set of 5-grams, and output to a file in your bucket on s3 (bigram analysis, though the words are non-contiguous).**\n",
    "\n",
    "2. **Using two (symmetric) comparison methods of your choice (e.g., correlations, distances, similarities), pairwise compare  all stripes (vectors), and output to a file in your bucket on s3.**\n",
    "\n",
    "> Design notes for (1)\n",
    "\n",
    "> For this task you will be able to modify the pattern we used in HW 3.2 (feel free to use the solution as reference). To total the word counts across the 5-grams, output the support from the mappers using the total order inversion pattern:\n",
    "\n",
    "    > <*word,count>\n",
    "\n",
    "> to ensure that the support arrives before the cooccurrences.\n",
    "\n",
    "> In addition to ensuring the determination of the total word counts, the mapper must also output co-occurrence counts for the pairs of words inside of each 5-gram. Treat these words as a basket, as we have in HW 3, but count all stripes or pairs in both orders, i.e., count both orderings: (word1,word2), and (word2,word1), to preserve symmetry in our output for (2).\n",
    "\n",
    "> Design notes for (2)\n",
    "\n",
    "> For this task you will have to determine a method of comparison. Here are a few that you might consider:\n",
    "\n",
    "> - Spearman correlation\n",
    "> - Euclidean distance\n",
    "> - Taxicab (Manhattan) distance\n",
    "> - Shortest path graph distance (a graph, because our data is symmetric!)\n",
    "> - Pearson correlation\n",
    "> - Cosine similarity\n",
    "> - Kendall correlation\n",
    "> - ...\n",
    "\n",
    "> However, be cautioned that some comparison methods are more difficult to parallelize than others, and do not perform more associations than is necessary, since your choice of association will be symmetric.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "I've chosen **Manhattan** (or **Taxicab**) **distance** (Euclideand distance could be implemented in the same way just calculating the square of all differences, and the square root of the sum of those squared differences), which is defined as:\n",
    "\n",
    "$$d_1(\\mathbf{p},\\mathbf{q})=\\left \\|\\mathbf{p}-\\mathbf{q}  \\right \\|_1=\n",
    "\\sum_{i=1}^N\\left | p_i-q_i \\right |$$"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##HW5.4.1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting HW541_TopN.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile HW541_TopN.py\n",
    "#!/home/hduser/anaconda/bin/python\n",
    "from mrjob.job import MRJob\n",
    "from mrjob.step import MRStep\n",
    "from mrjob.protocol import RawValueProtocol\n",
    "import re\n",
    "from operator import itemgetter\n",
    "from mrjob.compat import get_jobconf_value\n",
    "\n",
    "class HW541_TopN(MRJob):\n",
    "    \n",
    "    OUTPUT_PROTOCOL = RawValueProtocol\n",
    "    \n",
    "    def jobconf(self):\n",
    "        orig_jobconf = super(HW541_TopN, self).jobconf()        \n",
    "        custom_jobconf = {'mapred.reduce.tasks': '1'}\n",
    "        combined_jobconf = orig_jobconf\n",
    "        combined_jobconf.update(custom_jobconf)\n",
    "        self.jobconf = combined_jobconf\n",
    "        return combined_jobconf\n",
    "    \n",
    "    def configure_options(self):\n",
    "        super(HW541_TopN, self).configure_options()\n",
    "        # The number of most frequent unigrams can be configured by\n",
    "            # the user as an argument\n",
    "        self.add_passthrough_option('--number_unigrams',  \n",
    "                                    dest='number_unigrams', type='int', \n",
    "                                    default=10)\n",
    "    \n",
    "    def steps(self):\n",
    "        return [MRStep(mapper = self.mapper, combiner = self.combiner,\n",
    "                       reducer_init = self.reducer_init, \n",
    "                       reducer = self.reducer, \n",
    "                       reducer_final = self.reducer_final)]\n",
    "    \n",
    "    def mapper(self, _, line):\n",
    "        line.strip()\n",
    "        [ngram,count,pages,books] = re.split(\"\\t\",line)\n",
    "        # Output the count for each word in the 5-gram\n",
    "        unigrams = ngram.split()\n",
    "        for unigram in unigrams:\n",
    "            yield unigram, int(count)\n",
    "\n",
    "    def combiner(self, unigram, count):\n",
    "        yield unigram, sum(count)\n",
    "\n",
    "    def reducer_init(self):\n",
    "        self.top = {}\n",
    "\n",
    "    def reducer(self, unigram, count):\n",
    "        total = sum(count)\n",
    "        # If we have not exceeded max size of the dictionary yet\n",
    "        if len(self.top.keys()) < self.options.number_unigrams:\n",
    "            self.top[unigram] = total\n",
    "        # If exceeded, include new unigram only if more frequent that\n",
    "                # other previously stored\n",
    "        else:\n",
    "            if total > min(self.top.values()):\n",
    "                # Remove unigram not so frequent\n",
    "                self.top.pop(min(self.top, key = self.top.get))\n",
    "                # Add new unigram\n",
    "                self.top[unigram] = total\n",
    "    \n",
    "    def reducer_final(self):\n",
    "        for unigram in self.top.keys():\n",
    "            yield None,unigram+'\\t'+str(self.top[unigram])\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    HW541_TopN.run()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!chmod +x HW541_TopN.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Test the MRJob task:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "using configs in /home/hduser/.mrjob.conf\n",
      "creating tmp directory /tmp/HW541_TopN.hduser.20151013.223953.985203\n",
      "writing to /tmp/HW541_TopN.hduser.20151013.223953.985203/step-0-mapper_part-00000\n",
      "Counters from step 1:\n",
      "  (no counters found)\n",
      "writing to /tmp/HW541_TopN.hduser.20151013.223953.985203/step-0-mapper-sorted\n",
      "> sort /tmp/HW541_TopN.hduser.20151013.223953.985203/step-0-mapper_part-00000\n",
      "writing to /tmp/HW541_TopN.hduser.20151013.223953.985203/step-0-reducer_part-00000\n",
      "Counters from step 1:\n",
      "  (no counters found)\n",
      "Moving /tmp/HW541_TopN.hduser.20151013.223953.985203/step-0-reducer_part-00000 -> /tmp/HW541_TopN.hduser.20151013.223953.985203/output/part-00000\n",
      "Streaming final output from /tmp/HW541_TopN.hduser.20151013.223953.985203/output\n",
      "removing tmp directory /tmp/HW541_TopN.hduser.20151013.223953.985203\n",
      "I\t805344\n",
      "the\t633346\n",
      "of\t476762\n",
      "have\t411226\n"
     ]
    }
   ],
   "source": [
    "!./HW541_TopN.py gbooks_filtered_sample.txt --number_unigrams=4 | sort -k2 -rn > Top100_sample.txt\n",
    "!head -10 Top100_sample.txt\n",
    "!cut -f1 Top100_sample.txt | sort -n > /tmp/Top10kWords.txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting HW541_TopN_driver.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile HW541_TopN_driver.py\n",
    "#!/home/hduser/anaconda/bin/python\n",
    "from HW541_TopN import HW541_TopN\n",
    "\n",
    "import os\n",
    "\n",
    "mr_job = HW541_TopN(args=[\n",
    "        's3://filtered-5grams/', '-r', 'emr', \n",
    "        '--number_unigrams=10000',\n",
    "        '--output-dir=s3://ucb-mids-mls-juanjocarin/Top10k_output',\n",
    "        '--no-output'])\n",
    "\n",
    "with mr_job.make_runner() as runner: \n",
    "    runner.run()\n",
    "\n",
    "os.system(\"aws s3 cp s3://ucb-mids-mls-juanjocarin/Top10k_output/part-00000 \\\n",
    "    ./Top10k.txt\")\n",
    "os.system(\"aws s3 rm s3://ucb-mids-mls-juanjocarin/Top10k_output/part-00000\")\n",
    "os.system(\"aws s3 rm s3://ucb-mids-mls-juanjocarin/Top10k_output/_SUCCESS\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!chmod +x HW541_TopN_driver.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 228,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "download: s3://ucb-mids-mls-juanjocarin/Top10k_output/part-00000 to ./Top10k.txt\n",
      "delete: s3://ucb-mids-mls-juanjocarin/Top10k_output/part-00000\n",
      "delete: s3://ucb-mids-mls-juanjocarin/Top10k_output/_SUCCESS\n"
     ]
    }
   ],
   "source": [
    "!./HW541_TopN_driver.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Keep only the frequent words, not their counts:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 234,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!cut -f1 Top10k.txt | sort -n > Top10kWords.txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting HW541.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile HW541.py\n",
    "#!/home/hduser/anaconda/bin/python\n",
    "from mrjob.job import MRJob\n",
    "from mrjob.step import MRStep\n",
    "from mrjob.protocol import RawValueProtocol\n",
    "import re\n",
    "from itertools import combinations \n",
    "from operator import itemgetter\n",
    "\n",
    "class HW541(MRJob):\n",
    "\n",
    "    # I keep 2 lists of the Frequent Unigrams dictionary\n",
    "        # Otherwise the single gets duplicated when run locally\n",
    "    TopFrequentUnigramsM = []\n",
    "    TopFrequentUnigramsR = []\n",
    "\n",
    "    def jobconf(self):\n",
    "        orig_jobconf = super(HW541, self).jobconf()        \n",
    "        custom_jobconf = {'mapred.reduce.tasks': '1'}\n",
    "        combined_jobconf = orig_jobconf\n",
    "        combined_jobconf.update(custom_jobconf)\n",
    "        self.jobconf = combined_jobconf\n",
    "        return combined_jobconf\n",
    "    \n",
    "    def steps(self):\n",
    "        return [MRStep(mapper_init = self.mapper_init, \n",
    "                       mapper = self.mapper, combiner = self.combiner, \n",
    "                       reducer_init = self.reducer_init, \n",
    "                       reducer = self.reducer)]\n",
    "    \n",
    "    ## pull in the top occurring words dictionary here for the mapper\n",
    "    def mapper_init(self):\n",
    "        f = open(\"Top10kWords.txt\",\"r\")\n",
    "        for unigram in f:\n",
    "            unigram = unigram.strip()\n",
    "            self.TopFrequentUnigramsM.append(unigram)\n",
    "        \n",
    "    def mapper(self, _, line):\n",
    "        cooccur = {}\n",
    "        line.strip()\n",
    "        [ngram,count,pages,books] = re.split(\"\\t\",line)\n",
    "        # Output the count for each word in the 5-gram\n",
    "        unigrams = ngram.split()\n",
    "        # Get all of the 2-sets\n",
    "        combs = list(combinations(unigrams,2))\n",
    "        for combination in combs:\n",
    "            unigram1,unigram2 = combination\n",
    "            if unigram1 in self.TopFrequentUnigramsM and \\\n",
    "                unigram2 in self.TopFrequentUnigramsM:\n",
    "                    cooccur.setdefault(unigram1,{})\n",
    "                    cooccur[unigram1].setdefault(unigram2,0)\n",
    "                    cooccur[unigram1][unigram2] += int(count)\n",
    "                    cooccur.setdefault(unigram2,{})\n",
    "                    cooccur[unigram2].setdefault(unigram1,0)\n",
    "                    cooccur[unigram2][unigram1] += int(count)\n",
    "        for unigram1 in cooccur.keys():\n",
    "            yield unigram1, cooccur[unigram1]\n",
    "        \n",
    "    def combiner(self, unigram1, values):\n",
    "        cooccur = {}\n",
    "        for stripe in values:\n",
    "            for unigram2 in stripe.keys():\n",
    "                cooccur.setdefault(unigram2,0)\n",
    "                cooccur[unigram2] += stripe[unigram2]\n",
    "        yield unigram1, cooccur\n",
    "\n",
    "    def reducer_init(self):\n",
    "        f = open(\"Top10kWords.txt\",\"r\")\n",
    "        for word in f:\n",
    "            word = word.strip()\n",
    "            self.TopFrequentUnigramsR.append(word)\n",
    "    \n",
    "    def reducer(self, unigram1, values):\n",
    "        cooccur = {}\n",
    "        for stripe in values:\n",
    "            for unigram2 in stripe.keys():\n",
    "                cooccur.setdefault(unigram2,0)\n",
    "                cooccur[unigram2] += stripe[unigram2]\n",
    "        for unigram2 in self.TopFrequentUnigramsR:\n",
    "            cooccur.setdefault(unigram2,0)\n",
    "        yield unigram1,','.join([str(cooccur[unigram2]) for unigram2 in \\\n",
    "                                  sorted(self.TopFrequentUnigramsR)])\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    HW541.run()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!chmod +x HW541.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Test with the sample file:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "using configs in /home/hduser/.mrjob.conf\n",
      "creating tmp directory /tmp/HW541.hduser.20151013.224022.306420\n",
      "writing to /tmp/HW541.hduser.20151013.224022.306420/step-0-mapper_part-00000\n",
      "Counters from step 1:\n",
      "  (no counters found)\n",
      "writing to /tmp/HW541.hduser.20151013.224022.306420/step-0-mapper-sorted\n",
      "> sort /tmp/HW541.hduser.20151013.224022.306420/step-0-mapper_part-00000\n",
      "writing to /tmp/HW541.hduser.20151013.224022.306420/step-0-reducer_part-00000\n",
      "Counters from step 1:\n",
      "  (no counters found)\n",
      "Moving /tmp/HW541.hduser.20151013.224022.306420/step-0-reducer_part-00000 -> /tmp/HW541.hduser.20151013.224022.306420/output/part-00000\n",
      "Streaming final output from /tmp/HW541.hduser.20151013.224022.306420/output\n",
      "removing tmp directory /tmp/HW541.hduser.20151013.224022.306420\n",
      "\"I\"\t\"34302,393307,25203,86000\"\n",
      "\"have\"\t\"393307,130,3747,14896\"\n",
      "\"of\"\t\"25203,3747,31638,317731\"\n",
      "\"the\"\t\"86000,14896,317731,46498\"\n"
     ]
    }
   ],
   "source": [
    "!./HW541.py gbooks_filtered_sample.txt --file /tmp/Top10kWords.txt > /tmp/Stripes.txt\n",
    "!head -20 /tmp/Stripes.txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 245,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting HW541_driver.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile HW541_driver.py\n",
    "#!/home/hduser/anaconda/bin/python\n",
    "from HW541 import HW541\n",
    "\n",
    "import os\n",
    "\n",
    "mr_job = HW541(args=['s3://filtered-5grams/', '-r', 'emr',\n",
    "                     '--file=Top10kWords.txt',\n",
    "                     '--output-dir=s3://ucb-mids-mls-juanjocarin/Stripes_output',\n",
    "                     '--no-output'])\n",
    "\n",
    "with mr_job.make_runner() as runner: \n",
    "    runner.run()\n",
    "\n",
    "os.system(\"aws s3 cp s3://ucb-mids-mls-juanjocarin/Stripes_output/part-00000 \\\n",
    "    s3://ucb-mids-mls-juanjocarin/Stripes.txt\")\n",
    "os.system(\"aws s3 rm s3://ucb-mids-mls-juanjocarin/Stripes_output/part-00000\")\n",
    "os.system(\"aws s3 rm s3://ucb-mids-mls-juanjocarin/Stripes_output/_SUCCESS\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 246,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!chmod +x HW541_driver.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 247,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "copy: s3://ucb-mids-mls-juanjocarin/Stripes_output/part-00000 to s3://ucb-mids-mls-juanjocarin/Stripes.txt\n",
      "delete: s3://ucb-mids-mls-juanjocarin/Stripes_output/part-00000\n",
      "delete: s3://ucb-mids-mls-juanjocarin/Stripes_output/_SUCCESS\n"
     ]
    }
   ],
   "source": [
    "!./HW541_driver.py "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we have the 10,000 stripes, that we'll use in the last stage, in a S3 bucket."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##HW5.4.2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Due to time constraints (calculating the **Manhattan** distances in AWS took **28 hours** with 4 m1.medium instances... adn after that it crashed, we think due to lack of resources), we were only able to implement one metric, instead of two. Implementing **Euclidean** distance would be quite easy: we would just have to square differences and calculate the square root of the sum."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's suppose that our coordinates are (for simplicity I'm using integers, though I've finally used confidences, which are float numbers):\n",
    "\n",
    "$\\begin{pmatrix}\n",
    "7 & 8 & 5\\\\\n",
    "8 & 4 & 1\\\\ \n",
    "5 & 1 & 9\n",
    "\\end{pmatrix}$\n",
    "\n",
    "The (Manhattan) distance matrix is easy to calculate (and of course the elements in the diagonal will be null):\n",
    "\n",
    "$\\begin{pmatrix}\n",
    "0 & 9 & 13\\\\\n",
    "9 & 0 & 14\\\\ \n",
    "13 & 14 & 0\n",
    "\\end{pmatrix}$\n",
    "\n",
    "With the first row, \n",
    "$\\begin{pmatrix}\n",
    "7 & 8 & 5\n",
    "\\end{pmatrix}$, corresponding to the 1st component, we can calculate $\\mid p_1 - q_1 \\mid$ for all possible combinations of $\\mathbf{p}$ and $\\mathbf{q}$: $\\begin{pmatrix}\n",
    "0 & 1 & 2\n",
    "\\end{pmatrix}$ if $\\mathbf{q}$ is the unigram corresponding to the first column, $\\begin{pmatrix}\n",
    "1 & 0 & 3\n",
    "\\end{pmatrix}$ if $\\mathbf{q}$ is the unigram corresponding to the second column, and so on. If we proceed the same way for all rows, for the unigram in the first column we could obtain the following matrix:\n",
    "\n",
    "$\\begin{pmatrix}\n",
    "0 & 1 & 2\\\\\n",
    "0 & 4 & 7\\\\ \n",
    "0 & 4 & 4\n",
    "\\end{pmatrix}$\n",
    "\n",
    "The row-wise sum of this matrix corresponds to the first row in our distance matrix: $\\begin{pmatrix}\n",
    "0 & 9 & 13\n",
    "\\end{pmatrix}$, which give us the first component of the distance between the first unigram and itself, between that unigram and the second, and between the first unigram and the third."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting HW542.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile HW542.py\n",
    "#!/home/hduser/anaconda/bin/python\n",
    "from mrjob.job import MRJob\n",
    "from mrjob.step import MRStep\n",
    "import re\n",
    "from itertools import combinations \n",
    "from operator import itemgetter\n",
    "from math import sqrt\n",
    "from mrjob.protocol import RawValueProtocol\n",
    "\n",
    "class HW542(MRJob):\n",
    "\n",
    "    def jobconf(self):\n",
    "        orig_jobconf = super(HW542, self).jobconf()        \n",
    "        custom_jobconf = {'mapred.reduce.tasks': '1',\n",
    "                          'mapred.output.key.comparator.class': \n",
    "                          'org.apache.hadoop.mapred.lib.KeyFieldBasedComparator',\n",
    "                          'mapred.text.key.comparator.options': '-k1n'}\n",
    "        combined_jobconf = orig_jobconf\n",
    "        combined_jobconf.update(custom_jobconf)\n",
    "        self.jobconf = combined_jobconf\n",
    "        return combined_jobconf\n",
    "    \n",
    "    \n",
    "    OUTPUT_PROTOCOL = RawValueProtocol\n",
    "\n",
    "    def steps(self):\n",
    "        return [MRStep(\n",
    "                mapper = self.mapper, \n",
    "                reducer = self.reducer)]\n",
    "    #,\n",
    "    #            MRStep(\n",
    "    #            reducer = self.reducer_aggregation)]\n",
    "\n",
    "    \n",
    "    def mapper(self, _, line):\n",
    "        # i-th line (corresponding to i-th unigram from the top 10,000 \n",
    "            # frequent contains the i-th coordinates for all unigrams\n",
    "        line = re.sub('\\\"', '', line)\n",
    "        line = line.split()\n",
    "        unigram = line[0]\n",
    "        coords = line[1].split(',')\n",
    "        N = len(coords)\n",
    "        # We have N (=10,000) coordinates and points\n",
    "        # For each row (or vector) of N elements we're going to calculate N\n",
    "            # other vectors, by subtracting the 1st, second, ... N-th element\n",
    "            # and taking the absolute value\n",
    "        # We also need the unigram, because (since they were ordered \n",
    "            # alphabetically) it will allow us to detect the value of \"i\"\n",
    "        for i in range(len(coords)):\n",
    "            yield i, (unigram,[abs(int(coords[i])-int(x)) for x in coords])\n",
    "    \n",
    "    def reducer(self, row, values):\n",
    "        unigram = []\n",
    "        sum_coord = None\n",
    "        for value in values:\n",
    "            N = len(value[1])\n",
    "            if not sum_coord:\n",
    "                sum_coord = [0]*N\n",
    "            unigram.append(value[0])\n",
    "            #sum_coord += [s+int(v) for s,v in zip(sum_coord,value)]\n",
    "            sum_coord = [s+int(v) for s,v in zip(sum_coord,value[1])]\n",
    "        yield None,sorted(unigram)[row]+','+','.join([str(x) for x in sum_coord])\n",
    "        \n",
    "if __name__ == '__main__':\n",
    "    HW542.run()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!chmod +x HW542.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's continue our test with the confidence matrix of the sample file:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "using configs in /home/hduser/.mrjob.conf\n",
      "creating tmp directory /tmp/HW542.hduser.20151013.224417.152742\n",
      "writing to /tmp/HW542.hduser.20151013.224417.152742/step-0-mapper_part-00000\n",
      "Counters from step 1:\n",
      "  (no counters found)\n",
      "writing to /tmp/HW542.hduser.20151013.224417.152742/step-0-mapper-sorted\n",
      "> sort /tmp/HW542.hduser.20151013.224417.152742/step-0-mapper_part-00000\n",
      "writing to /tmp/HW542.hduser.20151013.224417.152742/step-0-reducer_part-00000\n",
      "Counters from step 1:\n",
      "  (no counters found)\n",
      "Moving /tmp/HW542.hduser.20151013.224417.152742/step-0-reducer_part-00000 -> /tmp/HW542.hduser.20151013.224417.152742/output/part-00000\n",
      "Streaming final output from /tmp/HW542.hduser.20151013.224417.152742/output\n",
      "I,0,844742,636825,762139\n",
      "have,844742,0,702447,667659\n",
      "of,636825,702447,0,629272\n",
      "the,762139,667659,629272,0\n",
      "removing tmp directory /tmp/HW542.hduser.20151013.224417.152742\n"
     ]
    }
   ],
   "source": [
    "!./HW542.py /tmp/Stripes.txt | head -20"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As expected, $\\text{distance}(\\text{unigram}_i,\\text{unigram}_i)=0.0 \\text{ } \\forall i \\in \\{1,N\\}$. Since the number of unigrams we are comparing in this example is so small, it's very easy to check that the code works perfectly."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting HW542_driver.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile HW542_driver.py\n",
    "#!/home/hduser/anaconda/bin/python\n",
    "from HW542 import HW542\n",
    "\n",
    "import os\n",
    "\n",
    "mr_job = HW542(args=['s3://ucb-mids-mls-juanjocarin/Stripes.txt', \n",
    "                     '-r', 'emr',\n",
    "                     '--output-dir=s3://ucb-mids-mls-juanjocarin/Manhattan_output2',\n",
    "                     '--no-output'])\n",
    "with mr_job.make_runner() as runner:\n",
    "    runner.run()\n",
    "#os.system(\"aws s3 cp s3://ucb-mids-mls-juanjocarin/Manhattan_output/part-00000 \\\n",
    "#    s3://ucb-mids-mls-juanjocarin/Manhattan_distances.txt\")\n",
    "#os.system(\"aws s3 rm s3://ucb-mids-mls-juanjocarin/Manhattan_output/part-00000\")\n",
    "#os.system(\"aws s3 rm s3://ucb-mids-mls-juanjocarin/Manhattan_output/_SUCCESS\")    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!chmod +x HW542_driver.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Unfortunately, we were not able to check whether the implementation works at scale or not (it didn't, at least with the limited resources I used)...\n",
    "\n",
    "<img src=\"./output.JPG\">\n",
    "\n",
    "**Syslog**:\n",
    "\n",
    "    2015-10-13 22:21:40,139 INFO org.apache.hadoop.streaming.StreamJob (main):  map 72%  reduce 0%\n",
    "    2015-10-13 22:22:04,187 INFO org.apache.hadoop.streaming.StreamJob (main):  map 100%  reduce 100%\n",
    "    2015-10-13 22:22:04,188 INFO org.apache.hadoop.streaming.StreamJob (main): To kill this job, run:\n",
    "    2015-10-13 22:22:04,188 INFO org.apache.hadoop.streaming.StreamJob (main): /home/hadoop/bin/hadoop job  -Dmapred.job.tracker=172.31.1.224:9001 -kill job_201510111752_0001\n",
    "    2015-10-13 22:22:04,189 INFO org.apache.hadoop.streaming.StreamJob (main): Tracking URL: http://ip-172-31-1-224.ec2.internal:9100/jobdetails.jsp?jobid=job_201510111752_0001\n",
    "    2015-10-13 22:22:04,189 ERROR org.apache.hadoop.streaming.StreamJob (main): Job not successful. Error: # of failed Reduce Tasks exceeded allowed limit. FailedCount: 1. LastFailedTask: task_201510111752_0001_r_000000\n",
    "    2015-10-13 22:22:04,189 INFO org.apache.hadoop.streaming.StreamJob (main): killJob...\n",
    "\n",
    "**Stderr**:\n",
    "\n",
    "    Streaming Command Failed!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!./HW542_driver.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#HW5.5"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "##See the other notebooks"
   ]
  }
 ],
 "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.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}