course1-big-data-esentials/c01w03_StopWordsTask2.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "Xsd0oermS80K"
   },
   "source": [
    "# Hadoop Streaming assignment 2: Stop Words"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "V_0wfvKjS80L"
   },
   "source": [
    "The purpose of this task is to improve the previous \"Word rating\" program. You have to calculate how many stop words are there in the input dataset. Stop words list is in `/datasets/stop_words_en.txt` file. \n",
    "\n",
    "Use Hadoop counters to compute the number of stop words and total words in the dataset. The result is the percentage of stop words in the entire dataset (without percent symbol).\n",
    "\n",
    "There are several points for this task:\n",
    "\n",
    "1) As an output, you have to get the percentage of stop words in the entire dataset without percent symbol (correct answer on sample dataset is `41.603`).\n",
    "\n",
    "2) As you can see in the Hadoop Streaming userguide \"you will need to use `-files` option to tell the framework to pack your executable files as a part of a job submission.\"\n",
    "\n",
    "3) Do not forget to redirect junk output to `/dev/null`.\n",
    "\n",
    "4) You may modify mappers/reducers from \"Word rating\" task and parse its output to get the answer on \"Stop Words\" task.\n",
    "\n",
    "5) You may use mapper/reducer to get `\"Stop Words\"` and `\"Total Words\"` amounts and redirect them to sys.stderr. After that you may redirect the output of MapReduce to the parsed function. In this function you may find rows correspond to these amounts and compute the percentage.\n",
    "\n",
    "Here you can find the draft for the main steps of the task. You can use other methods to get the solution."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "EpO7whLNS80M"
   },
   "source": [
    "## Step 1. Create the mapper.\n",
    "\n",
    "<b>Hint:</b> Create the mapper, which calculates Total word and Stop word amounts. You may redirect this information to sys.stderr. This will make it possible to parse these data on the next steps.\n",
    "\n",
    "Example of the redirections:\n",
    "\n",
    "`print >> sys.stderr, \"reporter:counter:Wiki stats,Total words,%d\" % count`\n",
    "\n",
    "Remember about the Distributed cache. If we add option `-files mapper.py,reducer.py,/datasets/stop_words_en.txt`, then `mapper.py, reducer.py` and `stop_words_en.txt` file will be in the same directory on the datanodes. Hence, it is necessary to use a relative path `stop_words_en.txt` from the mapper to access this txt file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "colab": {
     "autoexec": {
      "startup": false,
      "wait_interval": 0
     }
    },
    "colab_type": "code",
    "id": "KU56zkC0S80M"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting mapper.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile mapper.py\n",
    "\n",
    "\n",
    "import sys\n",
    "import re\n",
    "\n",
    "\n",
    "reload(sys)\n",
    "sys.setdefaultencoding('utf-8')  # required to convert to unicode\n",
    "\n",
    "path = 'stop_words_en.txt'\n",
    "\n",
    "# Your code for reading stop words here\n",
    "\n",
    "with open(path, \"r\") as f:\n",
    "    stop_words = f.read().splitlines()\n",
    "\n",
    "for line in sys.stdin:\n",
    "    try:\n",
    "        article_id, text = unicode(line.strip()).split('\\t', 1)\n",
    "    except ValueError as e:\n",
    "        continue\n",
    "\n",
    "    words = re.split(\"\\W*\\s+\\W*\", text, flags=re.UNICODE)\n",
    "        \n",
    "    # Your code for mapper here.\n",
    "    for word in words:\n",
    "        print >> sys.stderr, \"reporter:counter:Wiki stats,Total words,%d\" % 1\n",
    "        if word.lower() in stop_words:\n",
    "            print >> sys.stderr, \"reporter:counter:Wiki stats,Stop words,%d\" % 1\n",
    "        print(\"{}\\t{}\".format(word, 1))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "-3EM-tBES80O"
   },
   "source": [
    "## Step 2. Create the reducer.\n",
    "\n",
    "Create the reducer, which will accumulate the information after the mapper step. You may implement the combiner if you want. It can be useful from optimizing and speed up your computations (see the lectures from the Week 2 for more details)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "colab": {
     "autoexec": {
      "startup": false,
      "wait_interval": 0
     }
    },
    "colab_type": "code",
    "id": "yslvpwpfS80P"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting reducer.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile reducer.py\n",
    "\n",
    "# Your code for reducer here.\n",
    "import sys\n",
    "\n",
    "current_key = None\n",
    "total_words = 0\n",
    "\n",
    "for line in sys.stdin:\n",
    "    try:\n",
    "        key, count = line.strip().split('\\t', 1)\n",
    "        count = int(count)\n",
    "\n",
    "        if current_key != key:\n",
    "            \n",
    "            if current_key:\n",
    "                print(\"{}\\t{}\".format(current_key, total_words))\n",
    "            \n",
    "            total_words = 0\n",
    "            current_key = key\n",
    "            \n",
    "        total_words += count\n",
    "    \n",
    "    except ValueError as e:\n",
    "        continue\n",
    "    \n",
    "if current_key:\n",
    "    print(\"{}\\t{}\".format(current_key, total_words))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "MZQKrDecS80Q"
   },
   "source": [
    "## Step 3. Create the parsed function.\n",
    "\n",
    "<b>Hint:</b> Create the function, which will parse MapReduce sys.stderr for Total word and Stop word amounts.\n",
    "\n",
    "The `./counter_process.py` script should do the following:\n",
    "\n",
    "- parse hadoop logs from Stderr,\n",
    "\n",
    "- retrieve values of 2 user-defined counters,\n",
    "\n",
    "- compute percentage and output it into the stdout."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "colab": {
     "autoexec": {
      "startup": false,
      "wait_interval": 0
     }
    },
    "colab_type": "code",
    "id": "XptufhbMS80R"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting counter_process.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile counter_process.py\n",
    "\n",
    "#! /usr/bin/env python\n",
    "\n",
    "import sys\n",
    "\n",
    "# Your functions may be here.\n",
    "\n",
    "\n",
    "stop_words = 0\n",
    "total_words = 1\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    # Your code here.\n",
    "    \n",
    "    for line in sys.stdin:\n",
    "        try: \n",
    "            key, value = line.strip().split('=', 1)\n",
    "            \n",
    "            if key == sys.argv[1]: \n",
    "                stop_words = float(value)\n",
    "            elif key == sys.argv[2]: \n",
    "                total_words = float(value)\n",
    "        \n",
    "        except Exception as e:\n",
    "            continue\n",
    "            \n",
    "    print((stop_words / total_words) * 100)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "O_7XrxbKS80S"
   },
   "source": [
    "## Step 4. Bash commands\n",
    "\n",
    "<b> Hints: </b> \n",
    "\n",
    "1) If you want to redirect standard output to txt file you may use the following argument in yarn jar:\n",
    "\n",
    "```\n",
    "yarn ... \\\n",
    "  ... \\\n",
    "  -output ${OUT_DIR} > /dev/null 2> $LOGS\n",
    "```\n",
    "\n",
    "2) For printing the percentage of stop words in the entire dataset you may parse the MapReduce output. Parsed script may be written in Python code. \n",
    "\n",
    "To get the result you may use the UNIX pipe operator `|`. The output of the first command acts as an input to the second command (see lecture file-content-exploration-2 for more details).\n",
    "\n",
    "With this operator you may use command `cat` to redirect the output of MapReduce to ./counter_process.py with arguments, which correspond to the `\"Stop words\"` and `\"Total words\"` counters. Example is the following:\n",
    "\n",
    "`cat $LOGS | python ./counter_process.py \"Stop words\" \"Total words\"`\n",
    "\n",
    "Now something about Hadoop counters naming. \n",
    " - Built-in Hadoop counters usually have UPPER_CASE names. To make the grading system possible to distinguish your custom counters and system ones please use the following pattern for their naming: `[Aa]aaa...` (all except the first letters should be in lowercase);\n",
    " - Another points is how Hadoop sorts the counters. It sorts them lexicographically. Grading system reads your first counter as Stop words counter and the second as Total words. Please name you counters in such way that Hadoop set the Stop words counter before the Total words. \n",
    " \n",
    "E.g. \"Stop words\" and \"Total words\" names are Ok because they correspond both requirements.\n",
    "\n",
    "3) In Python code sys.argv is a list, which contains the command-line arguments passed to the script. The name of the script is in `sys.argv[0]`. Other arguments begin from `sys.argv[1]`.\n",
    "\n",
    "Hence, if you have two arguments, which you send from the Bash to your python script, you may use arguments in your script with the following command:\n",
    "\n",
    "`function(sys.argv[1], sys.argv[2])`\n",
    "\n",
    "4) Do not forget about printing your MapReduce output in the last cell. You may use the next command:\n",
    "\n",
    "`cat $LOGS >&2`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "colab": {
     "autoexec": {
      "startup": false,
      "wait_interval": 0
     }
    },
    "colab_type": "code",
    "id": "bG3omEqzS80S"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "41.60262201698447\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "18/11/15 08:48:11 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032\n",
      "18/11/15 08:48:11 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032\n",
      "18/11/15 08:48:12 INFO mapred.FileInputFormat: Total input files to process : 1\n",
      "18/11/15 08:48:12 INFO mapreduce.JobSubmitter: number of splits:2\n",
      "18/11/15 08:48:12 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1542058676835_0003\n",
      "18/11/15 08:48:12 INFO impl.YarnClientImpl: Submitted application application_1542058676835_0003\n",
      "18/11/15 08:48:12 INFO mapreduce.Job: The url to track the job: http://fae961296adc:8088/proxy/application_1542058676835_0003/\n",
      "18/11/15 08:48:12 INFO mapreduce.Job: Running job: job_1542058676835_0003\n",
      "18/11/15 08:48:18 INFO mapreduce.Job: Job job_1542058676835_0003 running in uber mode : false\n",
      "18/11/15 08:48:18 INFO mapreduce.Job:  map 0% reduce 0%\n",
      "18/11/15 08:48:34 INFO mapreduce.Job:  map 4% reduce 0%\n",
      "18/11/15 08:48:40 INFO mapreduce.Job:  map 6% reduce 0%\n",
      "18/11/15 08:48:47 INFO mapreduce.Job:  map 7% reduce 0%\n",
      "18/11/15 08:48:53 INFO mapreduce.Job:  map 9% reduce 0%\n",
      "18/11/15 08:48:59 INFO mapreduce.Job:  map 11% reduce 0%\n",
      "18/11/15 08:49:05 INFO mapreduce.Job:  map 13% reduce 0%\n",
      "18/11/15 08:49:11 INFO mapreduce.Job:  map 15% reduce 0%\n",
      "18/11/15 08:49:17 INFO mapreduce.Job:  map 16% reduce 0%\n",
      "18/11/15 08:49:23 INFO mapreduce.Job:  map 18% reduce 0%\n",
      "18/11/15 08:49:29 INFO mapreduce.Job:  map 20% reduce 0%\n",
      "18/11/15 08:49:35 INFO mapreduce.Job:  map 22% reduce 0%\n",
      "18/11/15 08:49:41 INFO mapreduce.Job:  map 24% reduce 0%\n",
      "18/11/15 08:49:47 INFO mapreduce.Job:  map 25% reduce 0%\n",
      "18/11/15 08:49:53 INFO mapreduce.Job:  map 27% reduce 0%\n",
      "18/11/15 08:49:59 INFO mapreduce.Job:  map 29% reduce 0%\n",
      "18/11/15 08:50:05 INFO mapreduce.Job:  map 31% reduce 0%\n",
      "18/11/15 08:50:11 INFO mapreduce.Job:  map 33% reduce 0%\n",
      "18/11/15 08:50:17 INFO mapreduce.Job:  map 35% reduce 0%\n",
      "18/11/15 08:50:23 INFO mapreduce.Job:  map 36% reduce 0%\n",
      "18/11/15 08:50:29 INFO mapreduce.Job:  map 38% reduce 0%\n",
      "18/11/15 08:50:35 INFO mapreduce.Job:  map 40% reduce 0%\n",
      "18/11/15 08:50:41 INFO mapreduce.Job:  map 42% reduce 0%\n",
      "18/11/15 08:50:47 INFO mapreduce.Job:  map 44% reduce 0%\n",
      "18/11/15 08:50:53 INFO mapreduce.Job:  map 45% reduce 0%\n",
      "18/11/15 08:50:59 INFO mapreduce.Job:  map 47% reduce 0%\n",
      "18/11/15 08:51:05 INFO mapreduce.Job:  map 49% reduce 0%\n",
      "18/11/15 08:51:11 INFO mapreduce.Job:  map 51% reduce 0%\n",
      "18/11/15 08:51:17 INFO mapreduce.Job:  map 53% reduce 0%\n",
      "18/11/15 08:51:23 INFO mapreduce.Job:  map 54% reduce 0%\n",
      "18/11/15 08:51:29 INFO mapreduce.Job:  map 56% reduce 0%\n",
      "18/11/15 08:51:35 INFO mapreduce.Job:  map 58% reduce 0%\n",
      "18/11/15 08:51:41 INFO mapreduce.Job:  map 60% reduce 0%\n",
      "18/11/15 08:51:47 INFO mapreduce.Job:  map 62% reduce 0%\n",
      "18/11/15 08:51:53 INFO mapreduce.Job:  map 64% reduce 0%\n",
      "18/11/15 08:51:59 INFO mapreduce.Job:  map 65% reduce 0%\n",
      "18/11/15 08:52:05 INFO mapreduce.Job:  map 67% reduce 0%\n",
      "18/11/15 08:52:09 INFO mapreduce.Job:  map 83% reduce 0%\n",
      "18/11/15 08:52:11 INFO mapreduce.Job:  map 100% reduce 0%\n",
      "18/11/15 08:52:15 INFO mapreduce.Job:  map 100% reduce 13%\n",
      "18/11/15 08:52:16 INFO mapreduce.Job:  map 100% reduce 25%\n",
      "18/11/15 08:52:17 INFO mapreduce.Job:  map 100% reduce 38%\n",
      "18/11/15 08:52:18 INFO mapreduce.Job:  map 100% reduce 50%\n",
      "18/11/15 08:52:19 INFO mapreduce.Job:  map 100% reduce 75%\n",
      "18/11/15 08:52:20 INFO mapreduce.Job:  map 100% reduce 88%\n",
      "18/11/15 08:52:21 INFO mapreduce.Job:  map 100% reduce 100%\n",
      "18/11/15 08:52:21 INFO mapreduce.Job: Job job_1542058676835_0003 completed successfully\n",
      "18/11/15 08:52:21 INFO mapreduce.Job: Counters: 52\n",
      "\tFile System Counters\n",
      "\t\tFILE: Number of bytes read=19935013\n",
      "\t\tFILE: Number of bytes written=31232197\n",
      "\t\tFILE: Number of read operations=0\n",
      "\t\tFILE: Number of large read operations=0\n",
      "\t\tFILE: Number of write operations=0\n",
      "\t\tHDFS: Number of bytes read=76874501\n",
      "\t\tHDFS: Number of bytes written=5884946\n",
      "\t\tHDFS: Number of read operations=30\n",
      "\t\tHDFS: Number of large read operations=0\n",
      "\t\tHDFS: Number of write operations=16\n",
      "\tJob Counters \n",
      "\t\tKilled reduce tasks=1\n",
      "\t\tLaunched map tasks=2\n",
      "\t\tLaunched reduce tasks=8\n",
      "\t\tData-local map tasks=2\n",
      "\t\tTotal time spent by all maps in occupied slots (ms)=459306\n",
      "\t\tTotal time spent by all reduces in occupied slots (ms)=25758\n",
      "\t\tTotal time spent by all map tasks (ms)=459306\n",
      "\t\tTotal time spent by all reduce tasks (ms)=25758\n",
      "\t\tTotal vcore-milliseconds taken by all map tasks=459306\n",
      "\t\tTotal vcore-milliseconds taken by all reduce tasks=25758\n",
      "\t\tTotal megabyte-milliseconds taken by all map tasks=470329344\n",
      "\t\tTotal megabyte-milliseconds taken by all reduce tasks=26376192\n",
      "\tMap-Reduce Framework\n",
      "\t\tMap input records=4100\n",
      "\t\tMap output records=11937375\n",
      "\t\tMap output bytes=97842522\n",
      "\t\tMap output materialized bytes=9943111\n",
      "\t\tInput split bytes=228\n",
      "\t\tCombine input records=11937375\n",
      "\t\tCombine output records=718673\n",
      "\t\tReduce input groups=474354\n",
      "\t\tReduce shuffle bytes=9943111\n",
      "\t\tReduce input records=718673\n",
      "\t\tReduce output records=474354\n",
      "\t\tSpilled Records=2156019\n",
      "\t\tShuffled Maps =16\n",
      "\t\tFailed Shuffles=0\n",
      "\t\tMerged Map outputs=16\n",
      "\t\tGC time elapsed (ms)=1740\n",
      "\t\tCPU time spent (ms)=678490\n",
      "\t\tPhysical memory (bytes) snapshot=2236502016\n",
      "\t\tVirtual memory (bytes) snapshot=20220243968\n",
      "\t\tTotal committed heap usage (bytes)=1525153792\n",
      "\tShuffle Errors\n",
      "\t\tBAD_ID=0\n",
      "\t\tCONNECTION=0\n",
      "\t\tIO_ERROR=0\n",
      "\t\tWRONG_LENGTH=0\n",
      "\t\tWRONG_MAP=0\n",
      "\t\tWRONG_REDUCE=0\n",
      "\tWiki stats\n",
      "\t\tStop words=4966261\n",
      "\t\tTotal words=11937375\n",
      "\tFile Input Format Counters \n",
      "\t\tBytes Read=76874273\n",
      "\tFile Output Format Counters \n",
      "\t\tBytes Written=5884946\n",
      "18/11/15 08:52:21 INFO streaming.StreamJob: Output directory: coursera_mr_task21542271689615848\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "\n",
    "OUT_DIR=\"coursera_mr_task2\"$(date +\"%s%6N\")\n",
    "NUM_REDUCERS=8\n",
    "LOGS=\"stderr_logs.txt\"\n",
    "\n",
    "# Stub code for your job\n",
    "\n",
    "rm ${LOGS} > /dev/null\n",
    "\n",
    "yarn jar /opt/cloudera/parcels/CDH/lib/hadoop-mapreduce/hadoop-streaming.jar \\\n",
    "    -D mapred.jab.name=\"Streaming stop words\" \\\n",
    "    -D mapreduce.job.reduces=${NUM_REDUCERS} \\\n",
    "    -files mapper.py,reducer.py,/datasets/stop_words_en.txt \\\n",
    "    -mapper \"python mapper.py\" \\\n",
    "    -combiner \"python reducer.py\" \\\n",
    "    -reducer \"python reducer.py\" \\\n",
    "    -input /data/wiki/en_articles_part \\\n",
    "    -output ${OUT_DIR} > /dev/null 2> $LOGS\n",
    " \n",
    "cat $LOGS | python ./counter_process.py \"Stop words\" \"Total words\"\n",
    "cat $LOGS >&2\n",
    "\n",
    "hdfs dfs -rm -r -skipTrash ${OUT_DIR} > /dev/null"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [],
   "default_view": {},
   "name": "702_to_students.ipynb",
   "provenance": [],
   "version": "0.3.2",
   "views": {}
  },
  "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}