w261/wk4/MIDS-W261-2015-HWK-Week04-Satpati.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#DATASCI W261: Machine Learning at Scale "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* **Sayantan Satpati**\n",
    "* **sayantan.satpati@ischool.berkeley.edu**\n",
    "* **W261**\n",
    "* **Week-4**\n",
    "* **Assignment-4**\n",
    "* **Date of Submission: 29-SEP-2015**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#  === Week 4: mrjob & k-means ==="
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## HW 4.0\n",
    "---\n",
    "\n",
    "***What is MrJob? How is it different to Hadoop MapReduce?***\n",
    "\n",
    "* mrjob helps in writing MapReduce jobs in Python 2.6+ and run them on several platforms. We can:\n",
    "\n",
    "    1. Write multi-step MapReduce jobs in pure Python\n",
    "    2. Test on your local machine\n",
    "    3. Run on a Hadoop cluster\n",
    "    4. Run in the cloud using Amazon Elastic MapReduce (EMR)\n",
    "    \n",
    "* mrjob is the easiest route to writing Python programs that run on Hadoop. mrjob helps in testing the code locally without installing Hadoop or run it on a cluster.\n",
    "\n",
    "* Additionally, mrjob has extensive integration with Amazon Elastic MapReduce. Once it is set up, it’s as easy to run your job in the cloud as it is to run it on the laptop.\n",
    "\n",
    "* Here are a number of features of mrjob that make writing MapReduce jobs easier:\n",
    "\n",
    "    1. Keep all MapReduce code for one job in a single class\n",
    "    2. Easily upload and install code and data dependencies at runtime\n",
    "    3. Switch input and output formats with a single line of code\n",
    "    4. Automatically download and parse error logs for Python tracebacks\n",
    "    5. Put command line filters before or after your Python code\n",
    "\n",
    "* mrjob is a wrapper over hadoop map/reduce streaming API(s). mrjob provides a consistent interface across every environment it supports. No matter whether you’re running locally, in the cloud, or on your own cluster, your Python code doesn’t change at all. mrjob is a framework that assists in submitting your job to the Hadoop job tracker and in running each individual step under Hadoop Streaming.\n",
    "\n",
    "\n",
    "***What are the mapper_final(), combiner_final(), reducer_final() methods? When are they called?***\n",
    "\n",
    "1. mapper_final(): Used to define an action to run after the mapper reaches the end of input.\n",
    "2. combiner_final(): Used to define an action to run after the combiner reaches the end of input.\n",
    "3. reducer_final(): Used to define an action to run after the reducer reaches the end of input.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## HW 4.1\n",
    "---\n",
    "\n",
    "***What is serialization in the context of MrJob or Hadoop?***\n",
    "\n",
    "* Serialization is the process of converting an object into a stream of bytes in order to store the object or transmit it to memory, a database, or a file. Its main purpose is to save the state of an object in order to be able to recreate it when needed. The reverse process is called deserialization.\n",
    "\n",
    "* In the context of Hadoop, serialization is process of converting the data into a compact binary format so that vast amounts of data can be quickly transferred over network, and be consumed by programs written in various lamguages. Instead of using Java Serialization Hadoop has its own serialization framework. The main perceived problems with Java Serialization was that it writes the classname of each object being serialized to the stream, with each subsequent instance of that class containing a 5 byte reference to the first, instead of the classname. This reduced the effective bandwidth and caused problems with random access and sorting of records in a serialiazed stream.\n",
    "\n",
    "***When it used in these frameworks?***\n",
    "\n",
    "* Serialization is used while reading and writing data before and after each step (map, combine, shuffle, reduce etc) when data is read from and written to disk and/or wire for transfer over the network. \n",
    "\n",
    "***What is the default serialization mode for input and outputs for MrJob?***\n",
    "\n",
    "* The default input protocol is RawValueProtocol, which reads and writes lines of raw text with no key. So by default, the first step in your job sees (None, <text of the line>) for each line of input.\n",
    "\n",
    "* The default output and internal protocols are both JSONProtocol, which reads and writes JSON strings separated by a tab character. (Hadoop Streaming uses the tab character to separate keys and values within one line when it sorts your data [1].)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## HW 4.2\n",
    "---\n",
    "\n",
    "Recall the Microsoft logfiles data from the async lecture. The logfiles are described are located at:\n",
    "\n",
    "https://kdd.ics.uci.edu/databases/msweb/msweb.html\n",
    "http://archive.ics.uci.edu/ml/machine-learning-databases/anonymous/\n",
    "\n",
    "This dataset records which areas (Vroots) of www.microsoft.com each user visited in a one-week timeframe in Feburary 1998.\n",
    "\n",
    " Here, you must preprocess the data on a single node (i.e., not on a cluster of nodes) from the format:\n",
    "\n",
    "```\n",
    "C,\"10001\",10001   #Visitor id 10001\n",
    "V,1000,1          #Visit by Visitor 10001 to page id 1000\n",
    "V,1001,1          #Visit by Visitor 10001 to page id 1001\n",
    "V,1002,1          #Visit by Visitor 10001 to page id 1002\n",
    "C,\"10002\",10002   #Visitor id 10001\n",
    "V\n",
    "Note: #denotes comments\n",
    "to the format:\n",
    "\n",
    "V,1000,1,C, 10001\n",
    "V,1001,1,C, 10001\n",
    "V,1002,1,C, 10001\n",
    "```\n",
    "\n",
    "Write the python code to accomplish this."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting preprocess_hw42.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile preprocess_hw42.py\n",
    "#!/usr/bin/env python\n",
    "import sys\n",
    "import os\n",
    "\n",
    "if len(sys.argv) < 2:\n",
    "    print \"No input file is passed, Aborting!!!\"\n",
    "    sys.exit(1)\n",
    "\n",
    "input_file = sys.argv[1]\n",
    "output_file = input_file + '.pp'\n",
    "\n",
    "try:\n",
    "    os.remove(output_file)\n",
    "except OSError:\n",
    "    pass\n",
    "\n",
    "last_visitor = None\n",
    "with open(input_file, 'r') as f1:\n",
    "    with open(output_file, 'a') as f2:\n",
    "        for line in f1:\n",
    "            line = line.strip()\n",
    "            tokens = line.split(\",\")\n",
    "            if len(tokens) == 3 and tokens[0] == 'C':\n",
    "                last_visitor = tokens[2]\n",
    "\n",
    "            if len(tokens) == 3 and tokens[0] == 'V':\n",
    "                out_line = '{0},C,{1}\\n'.format(line,last_visitor)\n",
    "                f2.write(out_line)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!chmod a+x preprocess_hw42.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "### Output (head)\n",
      "\n",
      "V,1000,1,C,10001\n",
      "V,1001,1,C,10001\n",
      "V,1002,1,C,10001\n",
      "V,1001,1,C,10002\n",
      "V,1003,1,C,10002\n",
      "V,1001,1,C,10003\n",
      "V,1003,1,C,10003\n",
      "V,1004,1,C,10003\n",
      "V,1005,1,C,10004\n",
      "V,1006,1,C,10005\n",
      "\n",
      "### Output (tail)\n",
      "\n",
      "V,1123,1,C,42708\n",
      "V,1038,1,C,42708\n",
      "V,1026,1,C,42708\n",
      "V,1041,1,C,42708\n",
      "V,1001,1,C,42709\n",
      "V,1003,1,C,42709\n",
      "V,1035,1,C,42710\n",
      "V,1001,1,C,42710\n",
      "V,1018,1,C,42710\n",
      "V,1008,1,C,42711\n"
     ]
    }
   ],
   "source": [
    "!python preprocess_hw42.py anonymous-msweb.data\n",
    "!echo \"### Output (head)\\n\"\n",
    "!head -n 10 anonymous-msweb.data.pp\n",
    "!echo \"\\n### Output (tail)\\n\"\n",
    "!tail -n 10 anonymous-msweb.data.pp\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##HW 4.3\n",
    "---\n",
    "\n",
    "**Find the 5 most frequently visited pages using mrjob from the output of 4.2 (i.e., transfromed log file).**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting mrjob_hw43.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile mrjob_hw43.py\n",
    "from mrjob.job import MRJob\n",
    "from mrjob.step import MRStep\n",
    "\n",
    "\n",
    "class MRVistedPagesCount(MRJob):\n",
    "    def steps(self):\n",
    "        return [\n",
    "            MRStep(mapper=self.mapper,\n",
    "                   combiner=self.combiner,\n",
    "                   reducer=self.reducer),\n",
    "            MRStep(reducer=self.reducer_find_top_5)\n",
    "        ]\n",
    "\n",
    "    def mapper(self, _, line):\n",
    "        tokens = line.strip().split(\",\")\n",
    "        yield tokens[1], 1\n",
    "\n",
    "    def combiner(self, page_visted, counts):\n",
    "        yield page_visted, sum(counts)\n",
    "\n",
    "    def reducer(self, page_visted, counts):\n",
    "        yield None, (sum(counts), page_visted)\n",
    "\n",
    "    # discard the key; it is just None\n",
    "    def reducer_find_top_5(self, _, page_visted_pairs):\n",
    "        # Store all the keys into memory (Assumption: Can be loaded into memory)\n",
    "        pairs = []\n",
    "        for p in page_visted_pairs:\n",
    "            pairs.append(p)\n",
    "        pairs.sort(key=lambda x: x[0], reverse=True)\n",
    "            \n",
    "        for p in pairs[:5]:\n",
    "            yield p[1],p[0]\n",
    "            \n",
    "\n",
    "if __name__ == '__main__':\n",
    "    MRVistedPagesCount.run()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!chmod a+x mrjob_hw43.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\"1008\"\t10836\r\n",
      "\"1034\"\t9383\r\n",
      "\"1004\"\t8463\r\n",
      "\"1018\"\t5330\r\n",
      "\"1017\"\t5108\r\n"
     ]
    }
   ],
   "source": [
    "# Running mrjob using command line\n",
    "!python mrjob_hw43.py -r local anonymous-msweb.data.pp -q"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING:mrjob.runner:\n",
      "WARNING:mrjob.runner:PLEASE NOTE: Starting in mrjob v0.5.0, protocols will be strict by default. It's recommended you run your job with --strict-protocols or set up mrjob.conf as described at https://pythonhosted.org/mrjob/whats-new.html#ready-for-strict-protocols\n",
      "WARNING:mrjob.runner:\n",
      "ERROR:mrjob.local:STDERR: + __mrjob_PWD=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/0\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python -c 'import fcntl; fcntl.flock(9, fcntl.LOCK_EX)'\n",
      "ERROR:mrjob.local:STDERR: + export PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/0/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/0/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + cd /tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/0\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python mrjob_hw43.py --step-num=0 --mapper /tmp/mrjob_hw43.cloudera.20150926.200443.882049/input_part-00000\n",
      "ERROR:mrjob.local:STDERR: + __mrjob_PWD=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/0\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python -c 'import fcntl; fcntl.flock(9, fcntl.LOCK_EX)'\n",
      "ERROR:mrjob.local:STDERR: + export PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/0/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/0/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + cd /tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/0\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python mrjob_hw43.py --step-num=0 --combiner\n",
      "ERROR:mrjob.local:STDERR: + __mrjob_PWD=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/1\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python -c 'import fcntl; fcntl.flock(9, fcntl.LOCK_EX)'\n",
      "ERROR:mrjob.local:STDERR: + export PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/1/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/1/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + cd /tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/1\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python mrjob_hw43.py --step-num=0 --mapper /tmp/mrjob_hw43.cloudera.20150926.200443.882049/input_part-00001\n",
      "ERROR:mrjob.local:STDERR: + __mrjob_PWD=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/1\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python -c 'import fcntl; fcntl.flock(9, fcntl.LOCK_EX)'\n",
      "ERROR:mrjob.local:STDERR: + export PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/1/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/1/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + cd /tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/mapper/1\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python mrjob_hw43.py --step-num=0 --combiner\n",
      "ERROR:mrjob.local:STDERR: + __mrjob_PWD=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/reducer/0\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python -c 'import fcntl; fcntl.flock(9, fcntl.LOCK_EX)'\n",
      "ERROR:mrjob.local:STDERR: + export PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/reducer/0/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/reducer/0/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + cd /tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/reducer/0\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python mrjob_hw43.py --step-num=0 --reducer /tmp/mrjob_hw43.cloudera.20150926.200443.882049/input_part-00000\n",
      "ERROR:mrjob.local:STDERR: + __mrjob_PWD=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/reducer/1\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python -c 'import fcntl; fcntl.flock(9, fcntl.LOCK_EX)'\n",
      "ERROR:mrjob.local:STDERR: + export PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/reducer/1/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/reducer/1/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + cd /tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/0/reducer/1\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python mrjob_hw43.py --step-num=0 --reducer /tmp/mrjob_hw43.cloudera.20150926.200443.882049/input_part-00001\n",
      "ERROR:mrjob.local:STDERR: + __mrjob_PWD=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/1/reducer/0\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python -c 'import fcntl; fcntl.flock(9, fcntl.LOCK_EX)'\n",
      "ERROR:mrjob.local:STDERR: + export PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/1/reducer/0/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + PYTHONPATH=/tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/1/reducer/0/mrjob.tar.gz:\n",
      "ERROR:mrjob.local:STDERR: + exec\n",
      "ERROR:mrjob.local:STDERR: + cd /tmp/mrjob_hw43.cloudera.20150926.200443.882049/job_local_dir/1/reducer/0\n",
      "ERROR:mrjob.local:STDERR: + /home/cloudera/anaconda/bin/python mrjob_hw43.py --step-num=1 --reducer /tmp/mrjob_hw43.cloudera.20150926.200443.882049/input_part-00000\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('1008', 10836)\n",
      "('1034', 9383)\n",
      "('1004', 8463)\n",
      "('1018', 5330)\n",
      "('1017', 5108)\n"
     ]
    }
   ],
   "source": [
    "# Running mrjob using a driver\n",
    "from mrjob_hw43 import MRVistedPagesCount\n",
    "mr_job = MRVistedPagesCount(args=['-r', 'local', 'anonymous-msweb.data.pp', 'q'])\n",
    "with mr_job.make_runner() as runner: \n",
    "    runner.run()\n",
    "    # stream_output: get access of the output \n",
    "    for line in runner.stream_output():\n",
    "        print mr_job.parse_output_line(line)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##HW 4.4 \n",
    "---\n",
    "\n",
    "**Find the most frequent visitor of each page using mrjob and the output of 4.2  (i.e., transfromed log file). In this output please include the webpage URL, webpageID and Visitor ID.**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Approach\n",
    "\n",
    "1. Create a separate file with URLS, i.e with records that start with A. This will be passed to the mrjob as an additional file for joining datasets\n",
    "2. Following happens in the first pass of mrjob:\n",
    "    * Mapper: Emits (pageId,visitorId), 1\n",
    "    * Combiner: Combines the counts\n",
    "    * Reducer: Combines the counts and emits records like (p1,v1) 100 | (p1, v2), 101 | (p1,v3), 202\n",
    "    * Partitioner is used to that all keys with the same pageId goes to the same reducer\n",
    "3. Following happens in the second pass of mrjob:\n",
    "    * Mapper: Emits (pageId,visit_count), visitorId\n",
    "    * Reducer_Init: Loads the url file into a dict\n",
    "    * Reducer: Emits (pageId, pageURL),(vists, vistorId)\n",
    "    * Partitioner is used to that all keys with the same pageId goes to the same reducer\n",
    "    * Secondary sorting (Descending Order) is done on the visit_count. This ensures that pages with highest page visits come first, before other records which has a lower page visit for the same page.\n",
    "   \n",
    "**Final Output Format(output_hw44.txt): (pageId, pageURL),(vists, vistorId)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "removed `url'\r\n"
     ]
    }
   ],
   "source": [
    "# Create a file with only URL(s), i.e. records starting with 'A'\n",
    "!rm -v url\n",
    "!grep ^A anonymous-msweb.data > url"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting mrjob_hw44.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile mrjob_hw44.py\n",
    "from mrjob.job import MRJob\n",
    "from mrjob.step import MRStep\n",
    "\n",
    "\n",
    "class MRFrequentVisitor(MRJob):\n",
    "    def steps(self):\n",
    "        return [\n",
    "            MRStep(mapper=self.mapper,\n",
    "                   combiner=self.combiner,\n",
    "                   reducer=self.reducer),\n",
    "             MRStep(mapper=self.mapper_frequent_visitor,\n",
    "                   reducer_init=self.reducer_frequent_visitor_init,\n",
    "                   reducer=self.reducer_frequent_visitor)\n",
    "        ]\n",
    "\n",
    "    def mapper(self, _, line):\n",
    "        tokens = line.strip().split(\",\")\n",
    "        key = \"{0},{1}\".format(tokens[1],tokens[4])\n",
    "        yield key, 1\n",
    "\n",
    "    def combiner(self, key, counts):\n",
    "        yield key, sum(counts)\n",
    "\n",
    "    def reducer(self, key, counts):\n",
    "        yield key, sum(counts)\n",
    "        \n",
    "    # 2nd Pass\n",
    "    \n",
    "    def mapper_frequent_visitor(self, key, value):\n",
    "        tokens = key.strip().split(\",\")\n",
    "        modified_key = \"{0},{1}\".format(tokens[0],value)\n",
    "        yield modified_key, tokens[1]\n",
    "     \n",
    "    \n",
    "    def reducer_frequent_visitor_init(self):\n",
    "        # Reads the 'url' file into a Dict for displaying additional information\n",
    "        self.last_page = None\n",
    "        self.pageDict = {}\n",
    "        with open('url','r') as f:\n",
    "            for line in f:\n",
    "                tokens = line.strip().split(\",\")\n",
    "                self.pageDict[tokens[1]] = tokens[4]\n",
    "                \n",
    "    def reducer_frequent_visitor(self, key, values):\n",
    "        tokens = key.strip().split(\",\")\n",
    "        page = tokens[0]\n",
    "        visits = int(tokens[1])\n",
    "        \n",
    "        if self.last_page != page:\n",
    "            self.last_page = page\n",
    "            # values might be a list, if there is a tie for same key => (p1, 1000), [v1,v2,v3..]\n",
    "            for value in values:\n",
    "                k = '{0},{1}'.format(page, \n",
    "                                    self.pageDict.get(page, 'NA').replace(\"\\\"\",\"\"))\n",
    "                v = '{0},{1}'.format(visits,\n",
    "                                    value)\n",
    "                yield k,v\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    MRFrequentVisitor.run()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!chmod a+x mrjob_hw44.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING:mrjob.runner:\n",
      "WARNING:mrjob.runner:PLEASE NOTE: Starting in mrjob v0.5.0, protocols will be strict by default. It's recommended you run your job with --strict-protocols or set up mrjob.conf as described at https://pythonhosted.org/mrjob/whats-new.html#ready-for-strict-protocols\n",
      "WARNING:mrjob.runner:\n",
      "WARNING:mrjob.compat:Detected hadoop configuration property names that do not match hadoop version 2.0.0:\n",
      "The have been translated as follows\n",
      " mapred.text.key.partitioner.options: mapreduce.partition.keypartitioner.options\n",
      "map.output.key.field.separator: mapreduce.map.output.key.field.separator\n",
      "mapred.output.key.comparator.class: mapreduce.job.output.key.comparator.class\n",
      "mapred.text.key.comparator.options: mapreduce.partition.keycomparator.options\n",
      "WARNING:mrjob.compat:Detected hadoop configuration property names that do not match hadoop version 2.0.0:\n",
      "The have been translated as follows\n",
      " mapred.text.key.partitioner.options: mapreduce.partition.keypartitioner.options\n",
      "map.output.key.field.separator: mapreduce.map.output.key.field.separator\n",
      "mapred.output.key.comparator.class: mapreduce.job.output.key.comparator.class\n",
      "mapred.text.key.comparator.options: mapreduce.partition.keycomparator.options\n"
     ]
    }
   ],
   "source": [
    "# Running mrjob using a Hadoop Runner in local cluster\n",
    "from mrjob_hw44 import MRFrequentVisitor\n",
    "import os\n",
    "\n",
    "# Passing Hadoop Streaming parameters to:\n",
    "# partition by leftmost part of composite key\n",
    "# secodary sort by rightmost part of the same composite key\n",
    "\n",
    "mr_job = MRFrequentVisitor(args=['-r', 'hadoop', \n",
    "                                 '--hadoop-home', '/usr/lib/hadoop-0.20-mapreduce',\n",
    "                                 '--hadoop-bin', '/usr/bin/hadoop',\n",
    "                                 '--file', 'url',\n",
    "                                 '--jobconf', 'stream.num.map.output.key.fields=2',\n",
    "                                 '--jobconf', 'map.output.key.field.separator=,',\n",
    "                                 '--jobconf', 'mapred.text.key.partitioner.options=-k1,1',\n",
    "                                 '--jobconf', 'mapred.text.key.comparator.options=-k1,1 -k2,2nr',\n",
    "                                 '--jobconf', 'mapred.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator',\n",
    "                                 '--partitioner', 'org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner',\n",
    "                                 'anonymous-msweb.data.pp', '-v'])\n",
    "\n",
    "\n",
    "output_file = \"output_hw44.txt\"\n",
    "try:\n",
    "    os.remove(output_file)\n",
    "except OSError:\n",
    "    pass\n",
    "\n",
    "with mr_job.make_runner() as runner, open(output_file, 'a') as f: \n",
    "    runner.run()\n",
    "    # stream_output: get access of the output \n",
    "    for line in runner.stream_output():\n",
    "        #print mr_job.parse_output_line(line)\n",
    "        f.write(line)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "   98654 output_hw44.txt\n",
      "\n",
      "### Output (head)\n",
      "\n",
      "\"1000,/regwiz\"\t\"1,42411\"\n",
      "\"1000,/regwiz\"\t\"1,42381\"\n",
      "\"1000,/regwiz\"\t\"1,42320\"\n",
      "\"1000,/regwiz\"\t\"1,42291\"\n",
      "\"1000,/regwiz\"\t\"1,42285\"\n",
      "\"1000,/regwiz\"\t\"1,42260\"\n",
      "\"1000,/regwiz\"\t\"1,42213\"\n",
      "\"1000,/regwiz\"\t\"1,42198\"\n",
      "\"1000,/regwiz\"\t\"1,42176\"\n",
      "\"1000,/regwiz\"\t\"1,42160\"\n",
      "\n",
      "### Output (tail)\n",
      "\n",
      "\"1295,/train_cert\"\t\"1,10345\"\n",
      "\"1295,/train_cert\"\t\"1,10340\"\n",
      "\"1295,/train_cert\"\t\"1,10325\"\n",
      "\"1295,/train_cert\"\t\"1,10316\"\n",
      "\"1295,/train_cert\"\t\"1,10271\"\n",
      "\"1295,/train_cert\"\t\"1,10208\"\n",
      "\"1295,/train_cert\"\t\"1,10205\"\n",
      "\"1295,/train_cert\"\t\"1,10204\"\n",
      "\"1295,/train_cert\"\t\"1,10090\"\n",
      "\"1295,/train_cert\"\t\"1,10028\"\n"
     ]
    }
   ],
   "source": [
    "# Output (Max page visit for all pages is 1; All ties have been reported)\n",
    "!wc -l output_hw44.txt\n",
    "!echo \"\\n### Output (head)\\n\"\n",
    "!head -n 10 output_hw44.txt \n",
    "!echo \"\\n### Output (tail)\\n\"\n",
    "!tail -n 10 output_hw44.txt "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##HW 4.5 \n",
    "---\n",
    "\n",
    "Here you will use a different dataset consisting of word-frequency distributions \n",
    "for 1,000 Twitter users. These Twitter users use language in very different ways,\n",
    "and were classified by hand according to the criteria:\n",
    "\n",
    "0: Human, where only basic human-human communication is observed.\n",
    "\n",
    "1: Cyborg, where language is primarily borrowed from other sources\n",
    "(e.g., jobs listings, classifieds postings, advertisements, etc...).\n",
    "\n",
    "2: Robot, where language is formulaically derived from unrelated sources\n",
    "(e.g., weather/seismology, police/fire event logs, etc...).\n",
    "\n",
    "3: Spammer, where language is replicated to high multiplicity\n",
    "(e.g., celebrity obsessions, personal promotion, etc... )\n",
    "\n",
    "Check out the preprints of our recent research,\n",
    "which spawned this dataset:\n",
    "\n",
    "http://arxiv.org/abs/1505.04342\n",
    "http://arxiv.org/abs/1508.01843\n",
    "\n",
    "The main data lie in the accompanying file:\n",
    "\n",
    "topUsers_Apr-Jul_2014_1000-words.txt\n",
    "\n",
    "and are of the form:\n",
    "\n",
    "USERID,CODE,TOTAL,WORD1_COUNT,WORD2_COUNT,...\n",
    ".\n",
    ".\n",
    "\n",
    "where\n",
    "\n",
    "USERID = unique user identifier\n",
    "CODE = 0/1/2/3 class code\n",
    "TOTAL = sum of the word counts\n",
    "\n",
    "Using this data, you will implement a 1000-dimensional K-means algorithm on the users\n",
    "by their 1000-dimensional word stripes/vectors using several \n",
    "centroid initializations and values of K.\n",
    "\n",
    "Note that each \"point\" is a user as represented by 1000 words, and that\n",
    "word-frequency distributions are generally heavy-tailed power-laws\n",
    "(often called Zipf distributions), and are very rare in the larger class\n",
    "of discrete, random distributions. For each user you will have to normalize\n",
    "by its \"TOTAL\" column. Try several parameterizations and initializations:\n",
    "\n",
    "(A) K=4 uniform random centroid-distributions over the 1000 words\n",
    "(B) K=2 perturbation-centroids, randomly perturbed from the aggregated (user-wide) distribution \n",
    "(C) K=4 perturbation-centroids, randomly perturbed from the aggregated (user-wide) distribution \n",
    "(D) K=4 \"trained\" centroids, determined by the sums across the classes.\n",
    "\n",
    "and iterate until a threshold (try 0.001) is reached.\n",
    "After convergence, print out a summary of the classes present in each cluster.\n",
    "In particular, report the composition as measured by the total\n",
    "portion of each class type (0-3) contained in each cluster,\n",
    "and discuss your findings and any differences in outcomes across parts A-D.\n",
    "\n",
    "Note that you do not have to compute the aggregated distribution or the \n",
    "class-aggregated distributions, which are rows in the auxiliary file:\n",
    "\n",
    "topUsers_Apr-Jul_2014_1000-words_summaries.txt"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Generate Initial Centroids for different use cases\n",
    "* (A) K=4 uniform random centroid-distributions over the 1000 words\n",
    "* (B) K=2 perturbation-centroids, randomly perturbed from the aggregated (user-wide) distribution \n",
    "* (C) K=4 perturbation-centroids, randomly perturbed from the aggregated (user-wide) distribution \n",
    "* (D) K=4 \"trained\" centroids, determined by the sums across the classes.\n",
    "\n",
    "### The following mrjob implements kmeans for 1000 dimensions and is common for each of the above use cases"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting mrjob_hw45.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile mrjob_hw45.py\n",
    "from numpy import argmin, array, random\n",
    "from mrjob.job import MRJob\n",
    "from mrjob.step import MRStep\n",
    "from mrjob.compat import get_jobconf_value \n",
    "from itertools import chain\n",
    "import sys\n",
    "\n",
    "#Calculate find the nearest centroid for data point \n",
    "def MinDist(datapoint, centroid_points):\n",
    "    datapoint = array(datapoint)\n",
    "    centroid_points = array(centroid_points)\n",
    "    diff = datapoint - centroid_points \n",
    "    diffsq = diff*diff\n",
    "    # Get the nearest centroid for each instance\n",
    "    minidx = argmin(list(diffsq.sum(axis = 1)))\n",
    "    return minidx\n",
    "\n",
    "#Check whether centroids converge\n",
    "def stop_criterion(centroid_points_old, centroid_points_new,T):\n",
    "    oldvalue = list(chain(*centroid_points_old))\n",
    "    newvalue = list(chain(*centroid_points_new))\n",
    "    Diff = [abs(x-y) for x, y in zip(oldvalue, newvalue)]\n",
    "    Flag = True\n",
    "    for i in Diff:\n",
    "        if(i>T):\n",
    "            Flag = False\n",
    "            break\n",
    "    return Flag\n",
    "\n",
    "class MRKmeans(MRJob):\n",
    "    centroid_points=[]\n",
    "    #k=0    \n",
    "    def steps(self):\n",
    "        return [\n",
    "            MRStep(mapper_init = self.mapper_init, mapper=self.mapper,combiner = self.combiner,reducer=self.reducer)\n",
    "               ]\n",
    "    #load centroids info from file\n",
    "    def mapper_init(self):\n",
    "        self.centroid_points = [map(float,s.split('\\n')[0].split(',')) for s in open(\"Centroids.txt\").readlines()]\n",
    "        open('Centroids.txt', 'w').close()\n",
    "    #load data and output the nearest centroid index and data point \n",
    "    def mapper(self, _, line):\n",
    "        D = (map(float,line.split(',')))\n",
    "        yield int(MinDist(D[3:],self.centroid_points)), (D[3:],1)\n",
    "    #Combine sum of data points locally\n",
    "    def combiner(self, idx, inputdata):\n",
    "        num = 0\n",
    "        sum_n = [0 for i in xrange(1000)]\n",
    "        for d, n in inputdata:\n",
    "            num = num + n\n",
    "            sum_n = [x + y for x,y in zip(d,sum_n)]\n",
    "        yield idx,(sum_n,num) \n",
    "    #Aggregate sum for each cluster and then calculate the new centroids\n",
    "    def reducer(self, idx, inputdata): \n",
    "        centroids = []\n",
    "        k = int(get_jobconf_value('k'))\n",
    "        num = [0] * k\n",
    "        for i in range(k):\n",
    "            centroids.append([0 for i in xrange(1000)])\n",
    "        for d, n in inputdata:\n",
    "            num[idx] = num[idx] + n\n",
    "            for i in xrange(1000):\n",
    "                centroids[idx][i] = centroids[idx][i] + d[i]\n",
    "        for i in xrange(1000):\n",
    "            centroids[idx][i] = centroids[idx][i]/num[idx]\n",
    "       \n",
    "        with open('Centroids.txt', 'a') as f:\n",
    "            f.writelines(\",\".join(str(i) for i in centroids[idx]) + '\\n')\n",
    "        yield idx,(centroids[idx], num)\n",
    "      \n",
    "if __name__ == '__main__':\n",
    "    MRKmeans.run()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!chmod a+x mrjob_hw45.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Pre-Process File By Normalizing the values in 'topUsers_Apr-Jul_2014_1000-words.txt'\n",
    "\n",
    "***Input: topUsers_Apr-Jul_2014_1000-words.txt***\n",
    "\n",
    "***Output: topUsers_Apr-Jul_2014_1000-words.txt.pp***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Min: 0.0\n",
      "Max: 0.40991\n"
     ]
    }
   ],
   "source": [
    "# Pre-Process the input by normalizing the values\n",
    "from sets import Set\n",
    "import os\n",
    "\n",
    "input_file = 'topUsers_Apr-Jul_2014_1000-words.txt'\n",
    "output_file = input_file + '.pp'\n",
    "\n",
    "try:\n",
    "    os.remove(output_file)\n",
    "except OSError:\n",
    "    pass\n",
    "\n",
    "values = Set()\n",
    "\n",
    "# Word Count has been normalized by the Total\n",
    "with open(input_file, 'r') as f1, open(output_file, 'w') as f2:\n",
    "    for line in f1:\n",
    "        count = 0\n",
    "        elements = line.strip().split(\",\")\n",
    "        total = int(elements[2])\n",
    "        for i in xrange(3, len(elements)):\n",
    "            value = round(float(elements[i]) * 1.0 / total, 6)\n",
    "            elements[i] = str(value)\n",
    "            values.add(value)\n",
    "        f2.write(\",\".join(elements) + \"\\n\")\n",
    "        \n",
    "print 'Min: {0}'.format(min(values))\n",
    "print 'Max: {0}'.format(max(values))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Generic Function with runner for running mrjob"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%load_ext autoreload\n",
    "%autoreload 2\n",
    "import numpy as np\n",
    "from numpy import random\n",
    "from mrjob_hw45 import MRKmeans, stop_criterion\n",
    "\n",
    "def kmeans(K, centroid_points):\n",
    "    mr_job = MRKmeans(args=['--file', output_file,\n",
    "                            '--jobconf', 'k={0}'.format(K),\n",
    "                            '--no-strict-protocol',\n",
    "                            'topUsers_Apr-Jul_2014_1000-words.txt.pp', '-v'])\n",
    "\n",
    "    # Update centroids iteratively\n",
    "    i = 0\n",
    "    while(1):\n",
    "        # save previous centoids to check convergency\n",
    "        centroid_points_old = centroid_points[:]\n",
    "        print \"iteration \" + str(i) + \":\"\n",
    "        with mr_job.make_runner() as runner: \n",
    "            runner.run()\n",
    "            # stream_output: get access of the output \n",
    "            for line in runner.stream_output():\n",
    "                key,value =  mr_job.parse_output_line(line)\n",
    "                centroid_points[key] = value[0]\n",
    "                print '[K={0}] Cluster IDX: {1}, \\\n",
    "                        Number of Elements: {2}, \\\n",
    "                        % of Elements: {3}'.format(K, key, \n",
    "                                                     value[1][key],\n",
    "                                                    value[1][key] * 100.0 / 1000)\n",
    "                #print key, len(value[0]), value[1]\n",
    "\n",
    "        print \"\\n\"\n",
    "        i = i + 1\n",
    "        if(stop_criterion(centroid_points_old,centroid_points,0.001)):\n",
    "            break\n",
    "    print \"--- Done ---\"\n",
    "    #print centroid_points"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Case: (A) K=4 uniform random centroid-distributions over the 1000 words\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING:mrjob.compat:get_jobconf_value() has been renamed to jobconf_from_env(). get_jobconf_value() will be removed in v0.5.0\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "iteration 0:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 649,                         % of Elements: 64.9\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 153,                         % of Elements: 15.3\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 34,                         % of Elements: 3.4\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 164,                         % of Elements: 16.4\n",
      "\n",
      "\n",
      "iteration 1:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 577,                         % of Elements: 57.7\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 252,                         % of Elements: 25.2\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 26,                         % of Elements: 2.6\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 145,                         % of Elements: 14.5\n",
      "\n",
      "\n",
      "iteration 2:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 617,                         % of Elements: 61.7\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 233,                         % of Elements: 23.3\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 20,                         % of Elements: 2.0\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 130,                         % of Elements: 13.0\n",
      "\n",
      "\n",
      "iteration 3:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 634,                         % of Elements: 63.4\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 219,                         % of Elements: 21.9\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 16,                         % of Elements: 1.6\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 131,                         % of Elements: 13.1\n",
      "\n",
      "\n",
      "iteration 4:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 677,                         % of Elements: 67.7\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 178,                         % of Elements: 17.8\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 14,                         % of Elements: 1.4\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 131,                         % of Elements: 13.1\n",
      "\n",
      "\n",
      "iteration 5:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 736,                         % of Elements: 73.6\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 121,                         % of Elements: 12.1\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 12,                         % of Elements: 1.2\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 131,                         % of Elements: 13.1\n",
      "\n",
      "\n",
      "iteration 6:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 767,                         % of Elements: 76.7\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 90,                         % of Elements: 9.0\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 12,                         % of Elements: 1.2\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 131,                         % of Elements: 13.1\n",
      "\n",
      "\n",
      "iteration 7:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 781,                         % of Elements: 78.1\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 76,                         % of Elements: 7.6\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 12,                         % of Elements: 1.2\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 131,                         % of Elements: 13.1\n",
      "\n",
      "\n",
      "iteration 8:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 788,                         % of Elements: 78.8\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 69,                         % of Elements: 6.9\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 12,                         % of Elements: 1.2\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 131,                         % of Elements: 13.1\n",
      "\n",
      "\n",
      "iteration 9:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 793,                         % of Elements: 79.3\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 64,                         % of Elements: 6.4\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 12,                         % of Elements: 1.2\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 131,                         % of Elements: 13.1\n",
      "\n",
      "\n",
      "iteration 10:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 794,                         % of Elements: 79.4\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 63,                         % of Elements: 6.3\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 12,                         % of Elements: 1.2\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 131,                         % of Elements: 13.1\n",
      "\n",
      "\n",
      "--- Done ---\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "\n",
    "K = 4\n",
    "output_file = 'Centroids.txt'\n",
    "\n",
    "try:\n",
    "    os.remove(output_file)\n",
    "except OSError:\n",
    "    pass\n",
    "\n",
    "centroid_points = []\n",
    "for k in xrange(K):\n",
    "    centroid_points.append(np.random.uniform(0.0, 0.40991, 1000)/1000)\n",
    "        \n",
    "with open(output_file, 'w+') as f:\n",
    "        f.writelines(','.join(str(j) for j in i) + '\\n' for i in centroid_points)\n",
    "        \n",
    "kmeans(K, centroid_points)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### (B) K=2 perturbation-centroids, randomly perturbed from the aggregated (user-wide) distribution \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0.999999999999999, 0.999999999999999]\n",
      "Standard Deviation 0.0024990758202\n",
      "Mean 0.001\n",
      "[1.0000000000000002, 1.0000000000000007]\n",
      "iteration 0:\n",
      "[K=2] Cluster IDX: 0,                         Number of Elements: 447,                         % of Elements: 44.7\n",
      "[K=2] Cluster IDX: 1,                         Number of Elements: 553,                         % of Elements: 55.3\n",
      "\n",
      "\n",
      "--- Done ---\n"
     ]
    }
   ],
   "source": [
    "K = 2\n",
    "output_file = 'Centroids.txt'\n",
    "\n",
    "try:\n",
    "    os.remove(output_file)\n",
    "except OSError:\n",
    "    pass\n",
    "\n",
    "centroid_points = []\n",
    "with open('topUsers_Apr-Jul_2014_1000-words_summaries.txt', 'r') as f:\n",
    "    for line in f:\n",
    "        if line.startswith('ALL_CODES'):\n",
    "            elements = line.strip().split(\",\")\n",
    "            total = int(elements[2])\n",
    "            array = [int(e) * 1.0 / total for e in elements[3:]]\n",
    "            for k in xrange(K):\n",
    "                centroid_points.append(array)\n",
    "            break;\n",
    "  \n",
    "print [sum(i) for i in centroid_points]\n",
    "std = np.std(centroid_points[0])\n",
    "print \"Standard Deviation\", std\n",
    "mean = np.mean(centroid_points[0])\n",
    "print \"Mean\", mean\n",
    "# Add Random Noise\n",
    "centroid_points = centroid_points + np.random.sample(K * 1000).reshape(K, 1000)\n",
    "for k in xrange(K):\n",
    "    # Normalize Again\n",
    "    centroid_points[k] = centroid_points[k] * 1.0 / np.sum(centroid_points[k])\n",
    "print [sum(i) for i in centroid_points]\n",
    "\n",
    "with open(output_file, 'w+') as f:\n",
    "        f.writelines(','.join(str(j) for j in i) + '\\n' for i in centroid_points)\n",
    "\n",
    "kmeans(K, centroid_points)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "### (C) K=4 perturbation-centroids, randomly perturbed from the aggregated (user-wide) distribution \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0.999999999999999, 0.999999999999999, 0.999999999999999, 0.999999999999999]\n",
      "Standard Deviation 0.0024990758202\n",
      "Mean 0.001\n",
      "[0.99999999999999989, 1.0000000000000018, 1.0000000000000004, 0.99999999999999889]\n",
      "iteration 0:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 544,                         % of Elements: 54.4\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 354,                         % of Elements: 35.4\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 99,                         % of Elements: 9.9\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 3,                         % of Elements: 0.3\n",
      "\n",
      "\n",
      "--- Done ---\n"
     ]
    }
   ],
   "source": [
    "K = 4\n",
    "output_file = 'Centroids.txt'\n",
    "\n",
    "try:\n",
    "    os.remove(output_file)\n",
    "except OSError:\n",
    "    pass\n",
    "\n",
    "centroid_points = []\n",
    "with open('topUsers_Apr-Jul_2014_1000-words_summaries.txt', 'r') as f:\n",
    "    for line in f:\n",
    "        if line.startswith('ALL_CODES'):\n",
    "            elements = line.strip().split(\",\")\n",
    "            total = int(elements[2])\n",
    "            array = [int(e) * 1.0 / total for e in elements[3:]]\n",
    "            for k in xrange(K):\n",
    "                centroid_points.append(array)\n",
    "            break;\n",
    "  \n",
    "print [sum(i) for i in centroid_points]\n",
    "std = np.std(centroid_points[0])\n",
    "print \"Standard Deviation\", std\n",
    "mean = np.mean(centroid_points[0])\n",
    "print \"Mean\", mean\n",
    "# Add Random Noise\n",
    "centroid_points = centroid_points + np.random.sample(K * 1000).reshape(K, 1000)\n",
    "for k in xrange(K):\n",
    "    # Normalize Again\n",
    "    centroid_points[k] = centroid_points[k] * 1.0 / np.sum(centroid_points[k])\n",
    "print [sum(i) for i in centroid_points]\n",
    "\n",
    "with open(output_file, 'w+') as f:\n",
    "        f.writelines(','.join(str(j) for j in i) + '\\n' for i in centroid_points)\n",
    "\n",
    "kmeans(K, centroid_points)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### (D) K=4 \"trained\" centroids, determined by the sums across the classes.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 245,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0.9999999999999996, 1.0000000000000002, 1.0, 1.0]\n",
      "iteration 0:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 787,                         % of Elements: 78.7\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 61,                         % of Elements: 6.1\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 82,                         % of Elements: 8.2\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 70,                         % of Elements: 7.0\n",
      "\n",
      "\n",
      "iteration 1:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 796,                         % of Elements: 79.6\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 54,                         % of Elements: 5.4\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 85,                         % of Elements: 8.5\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 65,                         % of Elements: 6.5\n",
      "\n",
      "\n",
      "iteration 2:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 802,                         % of Elements: 80.2\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 51,                         % of Elements: 5.1\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 84,                         % of Elements: 8.4\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 63,                         % of Elements: 6.3\n",
      "\n",
      "\n",
      "iteration 3:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 804,                         % of Elements: 80.4\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 51,                         % of Elements: 5.1\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 82,                         % of Elements: 8.2\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 63,                         % of Elements: 6.3\n",
      "\n",
      "\n",
      "iteration 4:\n",
      "[K=4] Cluster IDX: 0,                         Number of Elements: 804,                         % of Elements: 80.4\n",
      "[K=4] Cluster IDX: 1,                         Number of Elements: 51,                         % of Elements: 5.1\n",
      "[K=4] Cluster IDX: 2,                         Number of Elements: 82,                         % of Elements: 8.2\n",
      "[K=4] Cluster IDX: 3,                         Number of Elements: 63,                         % of Elements: 6.3\n",
      "\n",
      "\n",
      "--- Done ---\n"
     ]
    }
   ],
   "source": [
    "K = 4\n",
    "output_file = 'Centroids.txt'\n",
    "\n",
    "try:\n",
    "    os.remove(output_file)\n",
    "except OSError:\n",
    "    pass\n",
    "\n",
    "centroid_points = []\n",
    "with open('topUsers_Apr-Jul_2014_1000-words_summaries.txt', 'r') as f:\n",
    "    for line in f:\n",
    "        if line.startswith('CODE'):\n",
    "            elements = line.strip().split(\",\")\n",
    "            total = int(elements[2])\n",
    "            array = [int(e) * 1.0 / total for e in elements[3:]]\n",
    "            centroid_points.append(array)\n",
    "  \n",
    "print [sum(i) for i in centroid_points]\n",
    "\n",
    "with open(output_file, 'w+') as f:\n",
    "        f.writelines(','.join(str(j) for j in i) + '\\n' for i in centroid_points)\n",
    "\n",
    "kmeans(K, centroid_points)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "***The distributions are very similar in Case A & D with bulk of the users falling into 1 cluster (80%)***\n",
    "\n",
    "***In Case B, the users are much more evenly distributed***\n",
    "\n",
    "***In Case C, the users mostly fall into 2 clusters***"
   ]
  },
  {
   "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.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}