course1-big-data-esentials/c01w03_WordsRatingTask1.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "Ebk-YqgqEOw1"
   },
   "source": [
    "# Hadoop Streaming assignment 1: Words Rating"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "XDLjl7hGEOw2"
   },
   "source": [
    "The purpose of this task is to create your own WordCount program for Wikipedia dump processing and learn basic concepts of the MapReduce.\n",
    "\n",
    "In this task you have to find the 7th word by popularity and its quantity in the reverse order (most popular first) in Wikipedia data (`/data/wiki/en_articles_part`).\n",
    "\n",
    "There are several points for this task:\n",
    "\n",
    "1) As an output, you have to get the 7th word and its quantity separated by a tab character.\n",
    "\n",
    "2) You must use the second job to obtain a totally ordered result.\n",
    "\n",
    "3) Do not forget to redirect all trash and output to /dev/null.\n",
    "\n",
    "Here you can find the draft of the task main steps. You can use other methods for solution obtaining."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Instructions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create your own WordCount program and process Wikipedia dump. Use the second job to sort words by quantity in the reverse order (most popular first). Output format:\n",
    "\n",
    "    word <tab> count\n",
    "\n",
    "The result is the 7th word by popularity and its quantity.\n",
    "\n",
    "The result on the sample dataset:\n",
    "    \n",
    "    is  126420\n",
    "    \n",
    "**Hint**: it is possible to use exactly one reducer in the second job to obtain a totally ordered result."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you want to deploy the environment on your own machine, please use [bigdatateam/yarn-notebook](https://hub.docker.com/r/bigdatateam/yarn-notebook/) Docker container.   "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "lIUikVIDEOw2"
   },
   "source": [
    "## Step 1. Create mapper and reducer.\n",
    "\n",
    "<b>Hint:</b>  Demo task contains almost all the necessary pieces to complete this assignment. You may use the demo to implement the first MapReduce Job."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting mapper1.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile mapper1.py\n",
    "\n",
    "# Your code for mapper here.\n",
    "import sys\n",
    "import re\n",
    "\n",
    "reload(sys)\n",
    "sys.setdefaultencoding('utf-8') # required to convert to unicode\n",
    "\n",
    "for line in sys.stdin:\n",
    "    try:\n",
    "        article_id, text = unicode(line.strip()).split('\\t', 1)\n",
    "        words = re.split(\"\\W*\\s+\\W*\", text, flags=re.UNICODE)\n",
    "        \n",
    "        for word in words:\n",
    "            print \"%s\\t%d\" % (word.lower(), 1)\n",
    "\n",
    "    except ValueError as e:\n",
    "        continue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "colab": {
     "autoexec": {
      "startup": false,
      "wait_interval": 0
     }
    },
    "colab_type": "code",
    "id": "oE8KFKqpEOw5"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting reducer1.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile reducer1.py\n",
    "\n",
    "import sys\n",
    "\n",
    "current_word = None\n",
    "current_count = 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_word != key:\n",
    "            if current_word:\n",
    "                print \"%s\\t%d\" % (current_word, current_count)\n",
    "            \n",
    "            current_word = key\n",
    "            current_count = 0\n",
    "        \n",
    "        current_count += count\n",
    "        \n",
    "    except ValueError as e:\n",
    "        continue\n",
    "        \n",
    "if current_word:       \n",
    "    print \"%s\\t%d\" % (current_word, current_count)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "colab": {
     "autoexec": {
      "startup": false,
      "wait_interval": 0
     }
    },
    "colab_type": "code",
    "id": "LJtc3_u0uXSz"
   },
   "outputs": [],
   "source": [
    "# You can use this cell for other experiments: for example, for combiner."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "KYZpcwY9EOw6"
   },
   "source": [
    "## Step 2. Create sort job.\n",
    "\n",
    "<b>Hint:</b> You may use MapReduce comparator to solve this step. Make sure that the keys are sorted in ascending order."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "colab": {
     "autoexec": {
      "startup": false,
      "wait_interval": 0
     }
    },
    "colab_type": "code",
    "id": "g-NMzQ79EOw7"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting mapper2.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile mapper2.py\n",
    "\n",
    "import sys\n",
    "\n",
    "for line in sys.stdin:\n",
    "    try:\n",
    "        word, count = line.strip().split('\\t', 1)\n",
    "        count = int(count)\n",
    "        print(\"{0}\\t{1}\".format(count, word))\n",
    "    except ValueError as e:\n",
    "        continue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "colab": {
     "autoexec": {
      "startup": false,
      "wait_interval": 0
     }
    },
    "colab_type": "code",
    "id": "qs6SKs7wO4be"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting reducer2.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile reducer2.py\n",
    "\n",
    "import sys\n",
    "\n",
    "for line in sys.stdin:\n",
    "    try:\n",
    "        count, word = line.strip().split('\\t', 1)\n",
    "        count = int(count)\n",
    "        print(\"{0}\\t{1}\".format(word, count))\n",
    "    except ValueError as e:\n",
    "        continue"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "4OBE81GZEOw8"
   },
   "source": [
    "## Step 3. Bash commands\n",
    "\n",
    "<b> Hint: </b> For printing the exact row you may use basic UNIX commands. For instance, sed/head/tail/... (if you know other commands, you can use them).\n",
    "\n",
    "To run both jobs, you must use two consecutive yarn-commands. Remember that the input for the second job is the ouput for the first job."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "colab": {
     "autoexec": {
      "startup": false,
      "wait_interval": 0
     }
    },
    "colab_type": "code",
    "id": "1q9NczdIEOw9"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "is\t126420\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "rm: `assignment1_1_1541625507307963': No such file or directory\n",
      "18/11/07 21:18:30 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032\n",
      "18/11/07 21:18:31 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032\n",
      "18/11/07 21:18:31 INFO mapred.FileInputFormat: Total input files to process : 1\n",
      "18/11/07 21:18:31 INFO mapreduce.JobSubmitter: number of splits:2\n",
      "18/11/07 21:18:31 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1541617375890_0026\n",
      "18/11/07 21:18:31 INFO impl.YarnClientImpl: Submitted application application_1541617375890_0026\n",
      "18/11/07 21:18:31 INFO mapreduce.Job: The url to track the job: http://90e21216e8ac:8088/proxy/application_1541617375890_0026/\n",
      "18/11/07 21:18:31 INFO mapreduce.Job: Running job: job_1541617375890_0026\n",
      "18/11/07 21:18:37 INFO mapreduce.Job: Job job_1541617375890_0026 running in uber mode : false\n",
      "18/11/07 21:18:37 INFO mapreduce.Job:  map 0% reduce 0%\n",
      "18/11/07 21:18:54 INFO mapreduce.Job:  map 51% reduce 0%\n",
      "18/11/07 21:19:00 INFO mapreduce.Job:  map 71% reduce 0%\n",
      "18/11/07 21:19:02 INFO mapreduce.Job:  map 100% reduce 0%\n",
      "18/11/07 21:19:11 INFO mapreduce.Job:  map 100% reduce 13%\n",
      "18/11/07 21:19:12 INFO mapreduce.Job:  map 100% reduce 25%\n",
      "18/11/07 21:19:14 INFO mapreduce.Job:  map 100% reduce 38%\n",
      "18/11/07 21:19:15 INFO mapreduce.Job:  map 100% reduce 75%\n",
      "18/11/07 21:19:18 INFO mapreduce.Job:  map 100% reduce 100%\n",
      "18/11/07 21:19:18 INFO mapreduce.Job: Job job_1541617375890_0026 completed successfully\n",
      "18/11/07 21:19:18 INFO mapreduce.Job: Counters: 50\n",
      "\tFile System Counters\n",
      "\t\tFILE: Number of bytes read=243475435\n",
      "\t\tFILE: Number of bytes written=366548470\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=5370513\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)=44436\n",
      "\t\tTotal time spent by all reduces in occupied slots (ms)=49012\n",
      "\t\tTotal time spent by all map tasks (ms)=44436\n",
      "\t\tTotal time spent by all reduce tasks (ms)=49012\n",
      "\t\tTotal vcore-milliseconds taken by all map tasks=44436\n",
      "\t\tTotal vcore-milliseconds taken by all reduce tasks=49012\n",
      "\t\tTotal megabyte-milliseconds taken by all map tasks=45502464\n",
      "\t\tTotal megabyte-milliseconds taken by all reduce tasks=50188288\n",
      "\tMap-Reduce Framework\n",
      "\t\tMap input records=4100\n",
      "\t\tMap output records=11937375\n",
      "\t\tMap output bytes=97842436\n",
      "\t\tMap output materialized bytes=121717292\n",
      "\t\tInput split bytes=228\n",
      "\t\tCombine input records=0\n",
      "\t\tCombine output records=0\n",
      "\t\tReduce input groups=427176\n",
      "\t\tReduce shuffle bytes=121717292\n",
      "\t\tReduce input records=11937375\n",
      "\t\tReduce output records=427175\n",
      "\t\tSpilled Records=35812125\n",
      "\t\tShuffled Maps =16\n",
      "\t\tFailed Shuffles=0\n",
      "\t\tMerged Map outputs=16\n",
      "\t\tGC time elapsed (ms)=940\n",
      "\t\tCPU time spent (ms)=84640\n",
      "\t\tPhysical memory (bytes) snapshot=2291650560\n",
      "\t\tVirtual memory (bytes) snapshot=20238274560\n",
      "\t\tTotal committed heap usage (bytes)=1536163840\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",
      "\tFile Input Format Counters \n",
      "\t\tBytes Read=76874273\n",
      "\tFile Output Format Counters \n",
      "\t\tBytes Written=5370513\n",
      "18/11/07 21:19:18 INFO streaming.StreamJob: Output directory: assignment1_1_1541625507307963\n",
      "18/11/07 21:19:20 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032\n",
      "18/11/07 21:19:20 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032\n",
      "18/11/07 21:19:20 INFO mapred.FileInputFormat: Total input files to process : 8\n",
      "18/11/07 21:19:20 INFO mapreduce.JobSubmitter: number of splits:8\n",
      "18/11/07 21:19:21 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1541617375890_0027\n",
      "18/11/07 21:19:21 INFO impl.YarnClientImpl: Submitted application application_1541617375890_0027\n",
      "18/11/07 21:19:21 INFO mapreduce.Job: The url to track the job: http://90e21216e8ac:8088/proxy/application_1541617375890_0027/\n",
      "18/11/07 21:19:21 INFO mapreduce.Job: Running job: job_1541617375890_0027\n",
      "18/11/07 21:19:30 INFO mapreduce.Job: Job job_1541617375890_0027 running in uber mode : false\n",
      "18/11/07 21:19:30 INFO mapreduce.Job:  map 0% reduce 0%\n",
      "18/11/07 21:19:35 INFO mapreduce.Job:  map 75% reduce 0%\n",
      "18/11/07 21:19:38 INFO mapreduce.Job:  map 100% reduce 0%\n",
      "18/11/07 21:19:41 INFO mapreduce.Job:  map 100% reduce 100%\n",
      "18/11/07 21:19:42 INFO mapreduce.Job: Job job_1541617375890_0027 completed successfully\n",
      "18/11/07 21:19:42 INFO mapreduce.Job: Counters: 49\n",
      "\tFile System Counters\n",
      "\t\tFILE: Number of bytes read=6224888\n",
      "\t\tFILE: Number of bytes written=13710267\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=5371529\n",
      "\t\tHDFS: Number of bytes written=5370513\n",
      "\t\tHDFS: Number of read operations=27\n",
      "\t\tHDFS: Number of large read operations=0\n",
      "\t\tHDFS: Number of write operations=2\n",
      "\tJob Counters \n",
      "\t\tLaunched map tasks=8\n",
      "\t\tLaunched reduce tasks=1\n",
      "\t\tData-local map tasks=8\n",
      "\t\tTotal time spent by all maps in occupied slots (ms)=24033\n",
      "\t\tTotal time spent by all reduces in occupied slots (ms)=3603\n",
      "\t\tTotal time spent by all map tasks (ms)=24033\n",
      "\t\tTotal time spent by all reduce tasks (ms)=3603\n",
      "\t\tTotal vcore-milliseconds taken by all map tasks=24033\n",
      "\t\tTotal vcore-milliseconds taken by all reduce tasks=3603\n",
      "\t\tTotal megabyte-milliseconds taken by all map tasks=24609792\n",
      "\t\tTotal megabyte-milliseconds taken by all reduce tasks=3689472\n",
      "\tMap-Reduce Framework\n",
      "\t\tMap input records=427175\n",
      "\t\tMap output records=427175\n",
      "\t\tMap output bytes=5370522\n",
      "\t\tMap output materialized bytes=6224930\n",
      "\t\tInput split bytes=1016\n",
      "\t\tCombine input records=0\n",
      "\t\tCombine output records=0\n",
      "\t\tReduce input groups=2075\n",
      "\t\tReduce shuffle bytes=6224930\n",
      "\t\tReduce input records=427175\n",
      "\t\tReduce output records=427175\n",
      "\t\tSpilled Records=854350\n",
      "\t\tShuffled Maps =8\n",
      "\t\tFailed Shuffles=0\n",
      "\t\tMerged Map outputs=8\n",
      "\t\tGC time elapsed (ms)=966\n",
      "\t\tCPU time spent (ms)=16210\n",
      "\t\tPhysical memory (bytes) snapshot=2499387392\n",
      "\t\tVirtual memory (bytes) snapshot=18107703296\n",
      "\t\tTotal committed heap usage (bytes)=1753743360\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",
      "\tFile Input Format Counters \n",
      "\t\tBytes Read=5370513\n",
      "\tFile Output Format Counters \n",
      "\t\tBytes Written=5370513\n",
      "18/11/07 21:19:42 INFO streaming.StreamJob: Output directory: assignment1_2_1541625558696351\n",
      "cat: Unable to write to output stream.\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "\n",
    "OUT_DIR_1=\"assignment1_1_\"$(date +\"%s%6N\")\n",
    "NUM_REDUCERS=8\n",
    "\n",
    "hdfs dfs -rm -r -skipTrash ${OUT_DIR_1} > /dev/null\n",
    "\n",
    "# Code for your first job\n",
    "# yarn jar /opt/cloudera/parcels/CDH/lib/hadoop-mapreduce/hadoop-streaming.jar ...\n",
    "\n",
    "yarn jar /opt/cloudera/parcels/CDH/lib/hadoop-mapreduce/hadoop-streaming.jar \\\n",
    "    -D mapred.jab.name=\"Streaming wordCount\" \\\n",
    "    -D mapreduce.job.reduces=${NUM_REDUCERS} \\\n",
    "    -files mapper1.py,reducer1.py \\\n",
    "    -mapper \"python mapper1.py\" \\\n",
    "    -reducer \"python reducer1.py\" \\\n",
    "    -input /data/wiki/en_articles_part \\\n",
    "    -output ${OUT_DIR_1} > /dev/null\n",
    "\n",
    "# Code for your second job\n",
    "# yarn jar /opt/cloudera/parcels/CDH/lib/hadoop-mapreduce/hadoop-streaming.jar ...\n",
    "\n",
    "OUT_DIR_2=\"assignment1_2_\"$(date +\"%s%6N\")\n",
    "\n",
    "yarn jar /opt/cloudera/parcels/CDH/lib/hadoop-mapreduce/hadoop-streaming.jar \\\n",
    "    -D mapred.jab.name=\"Streaming wordCount 2\" \\\n",
    "    -D mapreduce.job.output.key.comparator.class=org.apache.hadoop.mapreduce.lib.partition.KeyFieldBasedComparator \\\n",
    "    -D mapreduce.partition.keycomparator.options=\"-k1,2nr\" \\\n",
    "    -files mapper2.py,reducer2.py \\\n",
    "    -mapper \"python mapper2.py\" \\\n",
    "    -reducer \"python reducer2.py\" \\\n",
    "    -numReduceTasks 1 \\\n",
    "    -input ${OUT_DIR_1} \\\n",
    "    -output ${OUT_DIR_2} > /dev/null\n",
    "\n",
    "# Code for obtaining the results\n",
    "hdfs dfs -cat ${OUT_DIR_2}/part-00000 | sed -n '7p;8q'\n",
    "hdfs dfs -rm -r -skipTrash ${OUT_DIR_1}* > /dev/null\n",
    "hdfs dfs -rm -r -skipTrash ${OUT_DIR_2}* > /dev/null"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [],
   "default_view": {},
   "name": "701_to_students.ipynb",
   "provenance": [],
   "version": "0.3.2",
   "views": {}
  },
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}