w261/wk3/MIDS-W261-2015-HWK-Week03-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-3**\n",
    "* **Assignment-4**\n",
    "* **Date of Submission: 22-SEP-2015**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#  === Week 3 Hadoop & Apriori ==="
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**What is a merge sort? Where is it used in Hadoop?**\n",
    "\n",
    "Merge sort is a comparison based sorting algorithm [O(nlogn)] which works by recursively dividing an unsorted array till the individual lists are of size 1, and then merging the sublists to produce new sorted sublists until there is only 1 sublist remaining.\n",
    "\n",
    "In Hadoop Merge Sort is used in the following places:\n",
    "Each mapper produces a sorted spill file (if it exceeds memory limitations) for each Reducer. Once the mappers are done processing on a single m/c, the spill files for each reducer are merged together by the combiner.\n",
    "Similarly, before the reduce phase, the sorted files from different mapper m/c(s) are merged together by a combiner before streaming the data to the reducer.\n",
    "\n",
    "**How is  a combiner function in the context of Hadoop?**\n",
    "\n",
    "Combiner is used in 1) Combine outputs from several mappers running on the same m/c, so that less amount of data is transferred over network & 2) Combine the outputs from different mapper m/c(s), before the data is streamed to the reducer.\n",
    "\n",
    "**Give an example where it can be used and justify why it should be used in the context of this problem.**\n",
    "\n",
    "For example in the Word Count problem, stop words (\"the\", \"for\", \"an\" etc) might occur very frequently. Instead of passing the tuple (\"the\", 1) 1 million times over network, the combiner acts an in-mapper mini reducer & combines the output of all the mappers running on the same m/c passing a single tuple (\"the\", 1000000), which  is much more efficient. However, there is a restriction - combiner can only be used when the function you want to apply to is both commutative (Ex: x + y = y + x), and associative (Ex: x + (y + z) = (x + y) + z). It can be used in Word Count problem since sum of the whole is equal to sum of its parts. But, it cannot be used to calculate something like a Median of a list of numbers.\n",
    "\n",
    "\n",
    "**What is the Hadoop shuffle?**\n",
    "\n",
    "Hadoop shuffle is the process of transferring data from mappers to reducers. It essentially involves **partition, sort, and combine** phases. Each mapper processes a chunk of the input data depending on the block size (64MB, 128 MB etc) and creates records for each Reducer (taking Partition Logic into consideration). These in-memory records might get written into disk as sorted spill files. Once all the mappers are done processing on a single m/c, the combiner may kick in and combine the data (multiple spill files) for each reducer using Merge Sort. Once all the mappers are done processing, the combined spill files (meant for each reducer) are transferred over network to the reducer node, where they are all Merge Sorted by the Combiner before streaming the sorted file to the Reducer.\n",
    "\n",
    "\n",
    "**What is the Apriori algorithm? Describe an example use in your domain of expertise. Define confidence and lift.**\n",
    "\n",
    "Per wikipedia, Apriori is an algorithm for frequent item set mining and association rule learning over transactional databases. It proceeds by identifying the frequent individual items in the database and extending them to larger and larger item sets as long as those item sets appear sufficiently often in the database. The frequent item sets determined by Apriori can be used to determine association rules which highlight general trends in the database: this has applications in domains such as market basket analysis.\n",
    "\n",
    "I work for ebay in the Search and Information Retrieval Space. Apriori Algorithm is used to power the related searches functionality in ebay. It shows related searches (based on keywords) to an user by mining the data on what other users might have searched when the searched using the same keyword that the current user has used.\n",
    "\n",
    "* **Confidence(X => Y)** : is the proportion the transactions that contains X, which also contains Y.\n",
    "* **Lift (X => Y)** : In general, if we have a high Support and a high Confidence, this means we have a strong association. However, this can be deceptive in some cases because if Item X and Item Y have a high support, we can have a high value for confidence even when they are independent. A better measure to judge the strength of an association rule is the Lift. The Lift ratio is the confidence of the rule divided by the confidence assuming the consequent (Item Y) and the antecedent (Item X) are independent"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### HW3.1.\n",
    "---\n",
    "Product Recommendations: The action or practice of selling additional products or services \n",
    "to existing customers is called cross-selling. Giving product recommendation is \n",
    "one of the examples of cross-selling that are frequently used by online retailers. \n",
    "One simple method to give product recommendations is to recommend products that are frequently\n",
    "browsed together by the customers.\n",
    "\n",
    "Suppose we want to recommend new products to the customer based on the products they\n",
    "have already browsed on the online website. Write a program using the A-priori algorithm\n",
    "to find products which are frequently browsed together. Fix the support to s = 100 \n",
    "(i.e. product pairs need to occur together at least 100 times to be considered frequent) \n",
    "and find itemsets of size 2 and 3.\n",
    "\n",
    "Use the online browsing behavior dataset at: \n",
    "\n",
    "https://www.dropbox.com/s/zlfyiwa70poqg74/ProductPurchaseData.txt?dl=0\n",
    "\n",
    "Each line in this dataset represents a browsing session of a customer. \n",
    "On each line, each string of 8 characters represents the id of an item browsed during that session. \n",
    "The items are separated by spaces.\n",
    "\n",
    "Do some exploratory data analysis of this dataset. \n",
    "Report your findings such as number of unique products; largest basket, etc. using Hadoop Map-Reduce."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "####  Environment \n",
    "\n",
    "##### Hadoop\n",
    "\n",
    "* For the purpose of this assignment, Cloudera VM (in Mac) has been used which comes pre-packaged with CDH 4.7.\n",
    "* Folder structure for input and output for HW3.1 and HW3.2\n",
    "\n",
    "```\n",
    "[cloudera@localhost wk3]$ hadoop fs -ls /user/cloudera/w261/wk3/*/input\n",
    "Found 1 items\n",
    "-rw-r--r--   3 cloudera cloudera    3458517 2015-09-21 13:15 /user/cloudera/w261/wk3/hw31/input/ProductPurchaseData.txt\n",
    "Found 1 items\n",
    "-rw-r--r--   3 cloudera cloudera    3458517 2015-09-21 13:15 /user/cloudera/w261/wk3/hw32/input/ProductPurchaseData.txt\n",
    "[cloudera@localhost wk3]$ hadoop fs -ls /user/cloudera/w261/wk3/*/output\n",
    "Found 3 items\n",
    "-rw-r--r--   3 cloudera cloudera          0 2015-09-21 13:16 /user/cloudera/w261/wk3/hw31/output/_SUCCESS\n",
    "drwxr-xr-x   - cloudera cloudera          0 2015-09-21 13:15 /user/cloudera/w261/wk3/hw31/output/_logs\n",
    "-rw-r--r--   3 cloudera cloudera        116 2015-09-21 13:16 /user/cloudera/w261/wk3/hw31/output/part-00000\n",
    "Found 3 items\n",
    "-rw-r--r--   3 cloudera cloudera          0 2015-09-21 13:18 /user/cloudera/w261/wk3/hw32/output/_SUCCESS\n",
    "drwxr-xr-x   - cloudera cloudera          0 2015-09-21 13:16 /user/cloudera/w261/wk3/hw32/output/_logs\n",
    "-rw-r--r--   3 cloudera cloudera        161 2015-09-21 13:18 /user/cloudera/w261/wk3/hw32/output/part-00000\n",
    "```\n",
    "\n",
    "##### Amazon EC2\n",
    "\n",
    "* For HW3.3, Ubuntu 1404 (Micro) VM was spinned up in Amazon EC2 Cluster (More details in that section) in order to run PyFim on the Product Data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting mapper_hw31.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile mapper_hw31.py\n",
    "\n",
    "#!/usr/bin/env python\n",
    "\n",
    "import sys\n",
    "import re\n",
    "import itertools\n",
    "\n",
    "for line in sys.stdin:\n",
    "    try:\n",
    "        # Remove leading & trailing chars\n",
    "        line = line.strip()\n",
    "        # Split the line by <TAB> delimeter\n",
    "        items = re.split(r'\\s', line)\n",
    "               \n",
    "        for i in items:\n",
    "            print '%s\\t%d\\t%d' %(i, 1, len(items))\n",
    "        \n",
    "        ''' Commenting out\n",
    "        #Sort the list\n",
    "        items.sort()\n",
    "        \n",
    "        for c in itertools.combinations(items, 1):\n",
    "            print '%s,%s\\t%d\\t%d' %(c[0], '*', 1, len(items))\n",
    "            \n",
    "        for c in itertools.combinations(items, 2):\n",
    "            print '%s,%s\\t%d' %(c[0], c[1], 1)\n",
    "        \n",
    "        for c in itertools.combinations(items, 3):\n",
    "            print '%s,%s,%s\\t%d' %(c[0], c[1], c[2], 1)\n",
    "        '''\n",
    "    except Exception as e:\n",
    "        print e"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!chmod a+x mapper_hw31.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting reducer_hw31.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile reducer_hw31.py\n",
    "#!/usr/bin/python\n",
    "import sys\n",
    "import re\n",
    "from sets import Set\n",
    "\n",
    "'''\n",
    "a1 1 5\n",
    "a1 1 5\n",
    "a1 1 6\n",
    "a2 1 18\n",
    "'''\n",
    "\n",
    "itemset_1_cnt = 0\n",
    "item_last = None\n",
    "\n",
    "THRESHOLD = 100\n",
    "\n",
    "# Statistics\n",
    "# Unique Items\n",
    "uniq = Set()\n",
    "# Max Basket Length\n",
    "max_basket_len = 0\n",
    "# Total Itemset Counts for Sizes: 1\n",
    "total_itemset_1 = 0\n",
    "\n",
    "for line in sys.stdin:\n",
    "    # Remove leading & trailing chars\n",
    "        line = line.strip()\n",
    "        # Split the line by <TAB> delimeter\n",
    "        tokens = re.split(r'\\s', line)\n",
    "    \n",
    "        item = tokens[0]\n",
    "        cnt = int(tokens[1])\n",
    "        basket_len = int(tokens[2])\n",
    "        \n",
    "        if not item_last:\n",
    "            item_last = item\n",
    "            \n",
    "        # Basket Length\n",
    "        if basket_len > max_basket_len:\n",
    "            max_basket_len = basket_len\n",
    "        \n",
    "        # Unique Items\n",
    "        uniq.add(item)\n",
    "        \n",
    "        if item_last != item:\n",
    "            # Check whether itemset 1 exceeds the support of 100\n",
    "            if itemset_1_cnt >= THRESHOLD:\n",
    "                total_itemset_1 += 1\n",
    "    \n",
    "            item_last = item\n",
    "            itemset_1_cnt = cnt\n",
    "        else:\n",
    "            itemset_1_cnt += cnt\n",
    "                    \n",
    "# Last Record\n",
    "if itemset_1_cnt >= THRESHOLD:\n",
    "    total_itemset_1 += 1\n",
    "\n",
    "print '=== Statistics ==='\n",
    "print 'Total Unique Items: %d' %(len(uniq))\n",
    "print 'Maximum Basket Length: %d' %(max_basket_len)\n",
    "print 'Total # frequent itemsets of size 1: %d' %(total_itemset_1)\n",
    "        \n",
    "        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!chmod a+x reducer_hw31.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Deleted /user/cloudera/w261/wk3/hw31/output\n",
      "packageJobJar: [./mapper_hw31.py, ./reducer_hw31.py, /tmp/hadoop-cloudera/hadoop-unjar1278596197593495009/] [] /tmp/streamjob4204464395455964255.jar tmpDir=null\n",
      "15/09/21 13:15:49 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.\n",
      "15/09/21 13:15:49 INFO mapred.FileInputFormat: Total input paths to process : 1\n",
      "15/09/21 13:15:50 INFO streaming.StreamJob: getLocalDirs(): [/tmp/hadoop-cloudera/mapred/local]\n",
      "15/09/21 13:15:50 INFO streaming.StreamJob: Running job: job_201509211304_0004\n",
      "15/09/21 13:15:50 INFO streaming.StreamJob: To kill this job, run:\n",
      "15/09/21 13:15:50 INFO streaming.StreamJob: UNDEF/bin/hadoop job  -Dmapred.job.tracker=localhost.localdomain:8021 -kill job_201509211304_0004\n",
      "15/09/21 13:15:50 INFO streaming.StreamJob: Tracking URL: http://0.0.0.0:50030/jobdetails.jsp?jobid=job_201509211304_0004\n",
      "15/09/21 13:15:51 INFO streaming.StreamJob:  map 0%  reduce 0%\n",
      "15/09/21 13:16:10 INFO streaming.StreamJob:  map 100%  reduce 0%\n",
      "15/09/21 13:16:23 INFO streaming.StreamJob:  map 100%  reduce 100%\n",
      "15/09/21 13:16:26 INFO streaming.StreamJob: Job complete: job_201509211304_0004\n",
      "15/09/21 13:16:26 INFO streaming.StreamJob: Output: /user/cloudera/w261/wk3/hw31/output\n",
      "=== Statistics ===\t\n",
      "Total Unique Items: 12592\t\n",
      "Maximum Basket Length: 37\t\n",
      "Total # frequent itemsets of size 1: 647\t\n"
     ]
    }
   ],
   "source": [
    "'''\n",
    "HW3.1. Product Recommendations\n",
    "'''\n",
    "\n",
    "# Delete existing Output Dirs if available\n",
    "!hadoop fs -rm -r -skipTrash /user/cloudera/w261/wk3/hw31/output\n",
    "\n",
    "# Run the Hadoop Streaming Command\n",
    "!hadoop jar /usr/lib/hadoop-0.20-mapreduce/contrib/streaming/hadoop-streaming-2.0.0-mr1-cdh4.7.0.jar \\\n",
    "-input /user/cloudera/w261/wk3/hw31/input/ProductPurchaseData.txt \\\n",
    "-output /user/cloudera/w261/wk3/hw31/output \\\n",
    "-file ./mapper_hw31.py \\\n",
    "-mapper 'python mapper_hw31.py' \\\n",
    "-file ./reducer_hw31.py \\\n",
    "-reducer 'python reducer_hw31.py'\n",
    "\n",
    "# Show Output\n",
    "!hadoop fs -cat /user/cloudera/w261/wk3/hw31/output/part-00000"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### HW3.2. (Computationally prohibitive but then again Hadoop can handle this)\n",
    "---\n",
    "\n",
    "Note: for this part the writeup will require a specific rule ordering but the program need not sort the output.\n",
    "\n",
    "List the top 5 rules with corresponding confidence scores in decreasing order of confidence score \n",
    "for frequent (100>count) itemsets of size 2. \n",
    "A rule is of the form: \n",
    "\n",
    "(item1) ⇒ item2.\n",
    "\n",
    "Fix the ordering of the rule lexicographically (left to right), \n",
    "and break ties in confidence (between rules, if any exist) \n",
    "by taking the first ones in lexicographically increasing order. \n",
    "Use Hadoop MapReduce to complete this part of the assignment; \n",
    "use a single mapper and single reducer; use a combiner if you think it will help and justify. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting mapper_hw32.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile mapper_hw32.py\n",
    "\n",
    "#!/usr/bin/env python\n",
    "\n",
    "import sys\n",
    "import re\n",
    "import itertools\n",
    "\n",
    "for line in sys.stdin:\n",
    "    try:\n",
    "        # Remove leading & trailing chars\n",
    "        line = line.strip()\n",
    "        # Split the line by <TAB> delimeter\n",
    "        items = re.split(r'\\s+', line)\n",
    "        #Sort the list\n",
    "        items.sort()\n",
    "        \n",
    "        l = len(items)\n",
    "        \n",
    "        for i in xrange(l):\n",
    "            print '%s,*\\t%d' %(items[i], 1)\n",
    "            for j in xrange(i+1, l):\n",
    "               print '%s,%s\\t%d' %(items[i], items[j], 1)\n",
    "               print '%s,%s\\t%d' %(items[j], items[i], 1)\n",
    "    except Exception as e:\n",
    "        print e"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!chmod a+x mapper_hw32.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting reducer_hw32.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile reducer_hw32.py\n",
    "#!/usr/bin/python\n",
    "import sys\n",
    "import re\n",
    "import heapq\n",
    "\n",
    "itemset_1_cnt = 0\n",
    "itemset_2_cnt = 0\n",
    "\n",
    "itemset_1_last = None\n",
    "itemset_2_last = None\n",
    "\n",
    "'''\n",
    "a1,* 1\n",
    "a1,* 1\n",
    "a1,b1 1\n",
    "a1,b1 1\n",
    "a1,b2 1\n",
    "a1,b2 1\n",
    "a2,* 1\n",
    "'''\n",
    "\n",
    "THRESHOLD = 100\n",
    "# Store Itemsets 2\n",
    "dict = {}\n",
    "\n",
    "for line in sys.stdin:\n",
    "    # Remove leading & trailing chars\n",
    "        line = line.strip()\n",
    "        # Split the line by <TAB> delimeter\n",
    "        tokens = re.split(r'\\s+', line)\n",
    "    \n",
    "        # Split the key by <COMMA> delimeter\n",
    "        items = tokens[0].split(\",\")\n",
    "        i1 = items[0]\n",
    "        i2 = items[1]\n",
    "        \n",
    "        if not itemset_1_last:\n",
    "            itemset_1_last = i1\n",
    "        \n",
    "        if itemset_1_last != i1:\n",
    "            '''\n",
    "            if itemset_1_cnt >= THRESHOLD:\n",
    "                confidence = (itemset_2_cnt * 1.0) / itemset_1_cnt\n",
    "                print '[%d,%d]%s\\t%f' %(itemset_1_cnt, itemset_2_cnt, tokens[0], confidence)\n",
    "                dict[tokens[0]] = confidence\n",
    "            '''\n",
    "                        \n",
    "            # Reset\n",
    "            itemset_1_last = i1\n",
    "            itemset_1_cnt = int(tokens[1])\n",
    "            itemset_2_last = None\n",
    "            itemset_2_cnt = 0\n",
    "        else:\n",
    "            if i2 == '*':\n",
    "                itemset_1_cnt += int(tokens[1])\n",
    "            else:\n",
    "                if itemset_2_last != tokens[0]:\n",
    "                    if itemset_1_cnt >= THRESHOLD and itemset_2_cnt >= THRESHOLD:\n",
    "                        confidence = (itemset_2_cnt * 1.0) / itemset_1_cnt\n",
    "                        #print '[%d,%d]%s\\t%f' %(itemset_1_cnt, itemset_2_cnt, itemset_2_last, confidence)\n",
    "                        dict[itemset_2_last] = confidence\n",
    "                    itemset_2_last = tokens[0]\n",
    "                    itemset_2_cnt = int(tokens[1]) \n",
    "                else:\n",
    "                    itemset_2_cnt += int(tokens[1])                    \n",
    "\n",
    "# Last Set of Counts\n",
    "if itemset_1_cnt >= THRESHOLD and itemset_2_cnt >= THRESHOLD:\n",
    "    confidence = (itemset_2_cnt * 1.0) / itemset_1_cnt\n",
    "    #print '[%d,%d]%s\\t%f' %(itemset_1_cnt, itemset_2_cnt, itemset_2_last, confidence)\n",
    "    dict[itemset_2_last] = confidence\n",
    "\n",
    "print '=== Top 5 Confidence ==='\n",
    "sorted_dict = sorted(dict.items(), key=lambda x:(-x[1], x[0]))\n",
    "for j,k in sorted_dict[:5]:\n",
    "    print '%s\\t%f' %(j,k)\n",
    "        \n",
    "        \n",
    "        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!chmod a+x reducer_hw32.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Deleted /user/cloudera/w261/wk3/hw32/output\n",
      "packageJobJar: [./mapper_hw32.py, ./reducer_hw32.py, /tmp/hadoop-cloudera/hadoop-unjar1867291528774777589/] [] /tmp/streamjob8875451858342305964.jar tmpDir=null\n",
      "15/09/21 13:16:53 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.\n",
      "15/09/21 13:16:53 INFO mapred.FileInputFormat: Total input paths to process : 1\n",
      "15/09/21 13:16:54 INFO streaming.StreamJob: getLocalDirs(): [/tmp/hadoop-cloudera/mapred/local]\n",
      "15/09/21 13:16:54 INFO streaming.StreamJob: Running job: job_201509211304_0005\n",
      "15/09/21 13:16:54 INFO streaming.StreamJob: To kill this job, run:\n",
      "15/09/21 13:16:54 INFO streaming.StreamJob: UNDEF/bin/hadoop job  -Dmapred.job.tracker=localhost.localdomain:8021 -kill job_201509211304_0005\n",
      "15/09/21 13:16:54 INFO streaming.StreamJob: Tracking URL: http://0.0.0.0:50030/jobdetails.jsp?jobid=job_201509211304_0005\n",
      "15/09/21 13:16:55 INFO streaming.StreamJob:  map 0%  reduce 0%\n",
      "15/09/21 13:17:11 INFO streaming.StreamJob:  map 19%  reduce 0%\n",
      "15/09/21 13:17:12 INFO streaming.StreamJob:  map 34%  reduce 0%\n",
      "15/09/21 13:17:14 INFO streaming.StreamJob:  map 53%  reduce 0%\n",
      "15/09/21 13:17:18 INFO streaming.StreamJob:  map 72%  reduce 0%\n",
      "15/09/21 13:17:21 INFO streaming.StreamJob:  map 91%  reduce 0%\n",
      "15/09/21 13:17:23 INFO streaming.StreamJob:  map 95%  reduce 0%\n",
      "15/09/21 13:17:24 INFO streaming.StreamJob:  map 100%  reduce 0%\n",
      "15/09/21 13:17:37 INFO streaming.StreamJob:  map 100%  reduce 68%\n",
      "15/09/21 13:17:44 INFO streaming.StreamJob:  map 100%  reduce 69%\n",
      "15/09/21 13:17:50 INFO streaming.StreamJob:  map 100%  reduce 70%\n",
      "15/09/21 13:17:56 INFO streaming.StreamJob:  map 100%  reduce 71%\n",
      "15/09/21 13:18:02 INFO streaming.StreamJob:  map 100%  reduce 72%\n",
      "15/09/21 13:18:04 INFO streaming.StreamJob:  map 100%  reduce 100%\n",
      "15/09/21 13:18:07 INFO streaming.StreamJob: Job complete: job_201509211304_0005\n",
      "15/09/21 13:18:07 INFO streaming.StreamJob: Output: /user/cloudera/w261/wk3/hw32/output\n",
      "=== Top 5 Confidence ===\t\n",
      "DAI93865,FRO40251\t1.000000\n",
      "GRO85051,FRO40251\t0.999176\n",
      "GRO38636,FRO40251\t0.990654\n",
      "ELE12951,FRO40251\t0.990566\n",
      "DAI88079,FRO40251\t0.986726\n"
     ]
    }
   ],
   "source": [
    "'''\n",
    "HW3.2. Confidence Caculations\n",
    "List the top 5 rules with corresponding confidence scores in decreasing order of confidence score \n",
    "for frequent (100>count) itemsets of size 2\n",
    "'''\n",
    "\n",
    "# Delete existing Output Dirs if available\n",
    "!hadoop fs -rm -r -skipTrash /user/cloudera/w261/wk3/hw32/output\n",
    "\n",
    "# Run the Hadoop Streaming Command\n",
    "!hadoop jar /usr/lib/hadoop-0.20-mapreduce/contrib/streaming/hadoop-streaming-2.0.0-mr1-cdh4.7.0.jar \\\n",
    "-D map.output.key.field.separator=, \\\n",
    "-D mapred.text.key.partitioner.options=-k1,1 \\\n",
    "-input /user/cloudera/w261/wk3/hw32/input/ProductPurchaseData.txt \\\n",
    "-output /user/cloudera/w261/wk3/hw32/output \\\n",
    "-file ./mapper_hw32.py \\\n",
    "-mapper 'python mapper_hw32.py' \\\n",
    "-file ./reducer_hw32.py \\\n",
    "-reducer 'python reducer_hw32.py'\n",
    "\n",
    "# Show Output\n",
    "!hadoop fs -cat /user/cloudera/w261/wk3/hw32/output/part-00000"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### HW3.3\n",
    "---\n",
    "\n",
    "Benchmark your results using the pyFIM implementation of the Apriori algorithm\n",
    "(Apriori - Association Rule Induction / Frequent Item Set Mining implemented by Christian Borgelt). \n",
    "You can download pyFIM from here:***\n",
    "\n",
    "http://www.borgelt.net/pyfim.html\n",
    "\n",
    "Comment on the results from both implementations (your Hadoop MapReduce of apriori versus pyFIM) \n",
    "in terms of results and execution times."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### For this part, the following steps were performed:\n",
    "\n",
    "1. Since I was using a Mac, I spinned up a Ubuntu 1404 (Micro) VM in Amazon EC2 Cluster\n",
    "2. Installed all required libraries: pip, git, ipython\n",
    "3. Downloaded the fim.so file from the link & set PYTHONPATH & LD_LIBRARY_PATH\n",
    "4. Ran the following command in the VM"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!python top5pyfim.py | sort -n -r -k 3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```\n",
    "FRO40251\t('DAI93865',)\t1.0\n",
    "FRO40251\t('GRO85051',)\t0.999176276771\n",
    "FRO40251\t('GRO38636',)\t0.990654205607\n",
    "FRO40251\t('ELE12951',)\t0.990566037736\n",
    "FRO40251\t('DAI88079',)\t0.986725663717\n",
    "FRO40251\t('FRO92469',)\t0.983510011779\n",
    "SNA82528\t('DAI43868',)\t0.972972972973\n",
    "DAI62779\t('DAI23334',)\t0.954545454545\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Top 5 Comparison from HW3.2 (Map/Reduce) and HW3.3 (PyFim Apriori)\n",
    "\n",
    "##### Map/Reduce\n",
    "```\n",
    "DAI93865,FRO40251\t1.000000\n",
    "GRO85051,FRO40251\t0.999176\n",
    "GRO38636,FRO40251\t0.990654\n",
    "ELE12951,FRO40251\t0.990566\n",
    "DAI88079,FRO40251\t0.986726\n",
    "```\n",
    "\n",
    "##### HW3.3 (PyFim Apriori)\n",
    "```\n",
    "FRO40251    ('DAI93865',)    1.0\n",
    "FRO40251    ('GRO85051',)    0.999176276771\n",
    "FRO40251    ('GRO38636',)    0.990654205607\n",
    "FRO40251    ('ELE12951',)    0.990566037736\n",
    "FRO40251    ('DAI88079',)    0.986725663717\n",
    "```\n",
    "**As it can be seen from above, both the results are identical with each other**\n",
    "\n",
    "* The Map/Reduce approach is much more scalable, even though it took longer to execute since there is a start up cost for Hadoop Map Reduce Jobs. As the file size gets bigger, the Map Reduce would perform much better than using PyFim."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### HW3.4 (Conceptual Exercise)\n",
    "\n",
    "Suppose that you wished to perform the Apriori algorithm once again,\n",
    "though this time now with the goal of listing the top 5 rules with corresponding confidence scores \n",
    "in decreasing order of confidence score for itemsets of size 3 using Hadoop MapReduce.\n",
    "A rule is now of the form: \n",
    "\n",
    "(item1, item2) ⇒ item3 \n",
    "\n",
    "Recall that the Apriori algorithm is iterative for increasing itemset size,\n",
    "working off of the frequent itemsets of the previous size to explore \n",
    "ONLY the NECESSARY subset of a large combinatorial space. \n",
    "Describe how you might design a framework to perform this exercise.\n",
    "\n",
    "In particular, focus on the following:\n",
    "  — map-reduce steps required\n",
    "  - enumeration of item sets and filtering for frequent candidates"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* In order to solve this problem a chain of Hadoop Map Reduce Jobs need to be used. The design would be something as follows:\n",
    "\n",
    "**Input Document: D**\n",
    "```\n",
    "item1 item2 item3 item4 item5\n",
    "item1 item2 item4 item5\n",
    "item3 item4 item5\n",
    ".\n",
    ".\n",
    ".\n",
    "```\n",
    "\n",
    "#### 1st Pass:\n",
    "  * **Mapper**: Read Document line by line and emit items as (item, 1)\n",
    "  * **Reducer**: Sum the items and emit only those that are > support count (Ex: s=100). At the end of the first pass the reducer file (L1) would look like the following. Since the output from Reducer is not supposed to be a big file, the number of reducer would be passed as 1 to the Hadoop Job, so that only one output file is generated from this pass.\n",
    "  ```\n",
    "    item1, 102\n",
    "    item3, 105\n",
    "  ```\n",
    "  \n",
    "#### 2nd Pass:\n",
    "  * **Mapper**: The output file from 1st pass would be passed to the Mapper in the form of a Distributed Cache. Assuming this file is small (Typical Super Store doesn't have more than 10,000 or so frequently sold items) it is loaded into the memory of the m/c(s) runnning mappers. The mapper would load L1 into a Dictionary (Ex: key=item1, value=102 etc). It would also create a candidate generation C2 by taking all combination pairs from L1 and create another Dictionary where keys are candidate pairs;values can be anyting like '-'. C2 would look like the following:\n",
    "  ```\n",
    "  (item1, item2) -\n",
    "  (item1, item3) -\n",
    "  ```\n",
    "  The mapper would then read portion of the Document D passed to it, and emit only those pairs which are present in C2 (Candidates). Output of mapper would have key like \"item1,item3\" and value as \"1\".\n",
    "  * **Reducer**: Hadoop framework would make sure Reducer receives the values for the same keys. The job of the reducer would be to sum the values for same keys, and emit only those that are > support count (Ex: s=100), just like in the 1st pass. Since the output from Reducer is not supposed to be a very big file (Though it will be much bigger than the file from 1st pass), the number of reducer would be passed as 1 to the Hadoop Job, so that only one output file is generated from this pass. The output of the reducer (L2) would look like the following. \n",
    "  ```\n",
    "  item1,item2 150\n",
    "  item1,item3 240\n",
    "  ```\n",
    "  \n",
    "#### 3rd Pass:\n",
    "  * **Mapper**: The output file from 2nd pass would be passed to the Mapper in the form of a Distributed Cache. Assuming this file is small it is loaded into the memory, as a Dictionary, of the m/c(s) runnning mappers. The mapper would read portion of the Document D passed to it, read line by line, and emit the following as long as the line has the candidate pairs. The mapper would emit something like the following if a candidate pair is present in the Document D:\n",
    "  ```\n",
    "  item1,item2,* 150\n",
    "  item1,item2,item4 1\n",
    "  ```\n",
    "  * **Reducer**: Paritioning has to be done based on the leftmost item pair (item1,item2), which would ensure all records with the same item pair would be passed to the same reducer. And secondary sort would ensure that the reducer would get to read the total (item1,item2,* 150) before it starts calculating the confidence scores. Each reducer would need to store the confidence scores in memory to be able to spit out the top 5 once the processing is done. If the number of records exceeds the memory of the reducer, then top 5 has to be recomputed every time a confidence score is calculated by the reducer."
   ]
  }
 ],
 "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
}