course1-big-data-esentials/c01w03_NameCountTask3.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Hadoop Streaming assignment 3: Name Count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Make WordCount program for all the names in the dataset. Name is a word with the following properties:\n",
    "\n",
    "1. The first character is not a digit (other characters can be digits).\n",
    "2. The first character is uppercase, all the other characters that are letters are lowercase.\n",
    "3. There are less than 0.5% occurrences of this word, when this word regardless to its case appears in the dataset and the condition (2) is not met.\n",
    "\n",
    "Order by quantity, most popular first, output format:\n",
    "\n",
    "    name <tab> count\n",
    "\n",
    "The result is the 5th line in the output.\n",
    "\n",
    "The result on the sample dataset:\n",
    "\n",
    "    french 5742\n",
    "    \n",
    "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": {},
   "source": [
    "## Step 1. Create first mapper and reducer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting mapper1.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile mapper1.py\n",
    "\n",
    "import sys\n",
    "import re\n",
    "\n",
    "for line in sys.stdin:\n",
    "\n",
    "    article_id, text = line.strip().split('\\t', 1)\n",
    "\n",
    "    try:\n",
    "        words = re.split('\\W*\\s+\\W*', text.strip())\n",
    "        for word in words:\n",
    "            cond1 = word[0].isalpha()\n",
    "            cond2 = not word[0].islower() and word[1:].islower()\n",
    "\n",
    "            if cond1:\n",
    "                print(\"{}\\t{:d}\\t{:d}\".format(word.lower(), 1, int(cond2)))\n",
    "            \n",
    "    except Exception as e:\n",
    "        continue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting reducer1.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile reducer1.py\n",
    "\n",
    "import sys\n",
    "\n",
    "current_key = None\n",
    "word_total = 0\n",
    "name_total = 0\n",
    "\n",
    "for line in sys.stdin:\n",
    "    try:\n",
    "        key, word_count, name_count = line.strip().split('\\t', 2)\n",
    "        word_count = int(word_count)\n",
    "        name_count = int(name_count)\n",
    "        \n",
    "        if current_key != key:\n",
    "            if current_key:\n",
    "                print(\"{}\\t{:d}\\t{:d}\".format(current_key, name_total, word_total))\n",
    "                \n",
    "            current_key = key\n",
    "            word_total = 0\n",
    "            name_total = 0\n",
    "        \n",
    "        word_total += word_count\n",
    "        name_total += name_count\n",
    "        \n",
    "    except Exception as e:\n",
    "        print(e)\n",
    "        continue  \n",
    "\n",
    "if current_key:\n",
    "    print(\"{}\\t{:d}\\t{:d}\".format(current_key, name_total, word_total))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2. Create second mapper and reducer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting mapper2.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile mapper2.py\n",
    "\n",
    "import sys\n",
    "\n",
    "current_key = None\n",
    "word_total = 0\n",
    "name_total = 0\n",
    "\n",
    "for line in sys.stdin:\n",
    "    try:\n",
    "        key, name_count, word_count = line.strip().split('\\t', 2)\n",
    "        word_count = int(word_count)\n",
    "        name_count = int(name_count)\n",
    "        \n",
    "        if current_key != key:\n",
    "            if current_key:\n",
    "                print(\"{}\\t{}\\t{}\".format(name_total, word_total, current_key))\n",
    "                \n",
    "            current_key = key\n",
    "            word_total = 0\n",
    "            name_total = 0\n",
    "        \n",
    "        word_total += word_count\n",
    "        name_total += name_count\n",
    "        \n",
    "    except ValueError as e:\n",
    "        print(e)\n",
    "        continue    \n",
    "\n",
    "if current_key:\n",
    "    print(\"{}\\t{}\\t{}\".format(name_total, word_total, current_key))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting reducer2.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile reducer2.py\n",
    "\n",
    "import sys\n",
    "\n",
    "total_count = 0\n",
    "total_caps = 0\n",
    "current_word = None\n",
    "\n",
    "for line in sys.stdin:\n",
    "    try:\n",
    "        caps_count, count, key = line.strip().split('\\t', 2)\n",
    "        count = int(count)\n",
    "        caps_count = int(caps_count)\n",
    "        \n",
    "        if key != current_word:\n",
    "            \n",
    "            if current_word and float(total_caps) / float(total_count) >= 0.995: \n",
    "                print(\"{}\\t{:d}\".format(current_word, total_caps))\n",
    "            \n",
    "            total_count = 0\n",
    "            total_caps = 0\n",
    "            current_word = key\n",
    "\n",
    "\n",
    "        total_caps += caps_count\n",
    "        total_count += count\n",
    "\n",
    "    except Exception as e:\n",
    "        print e\n",
    "        continue\n",
    "        \n",
    "if current_word and float(total_caps) / float(total_count) >= 0.995: \n",
    "    print(\"{}\\t{:d}\".format(current_word, total_caps))   "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "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": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "french\t5740\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "18/11/08 22:14:24 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032\n",
      "18/11/08 22:14:24 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032\n",
      "18/11/08 22:14:25 INFO mapred.FileInputFormat: Total input files to process : 1\n",
      "18/11/08 22:14:25 INFO mapreduce.JobSubmitter: number of splits:2\n",
      "18/11/08 22:14:25 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1541700195986_0073\n",
      "18/11/08 22:14:25 INFO impl.YarnClientImpl: Submitted application application_1541700195986_0073\n",
      "18/11/08 22:14:25 INFO mapreduce.Job: The url to track the job: http://f3f7668c3905:8088/proxy/application_1541700195986_0073/\n",
      "18/11/08 22:14:25 INFO mapreduce.Job: Running job: job_1541700195986_0073\n",
      "18/11/08 22:14:31 INFO mapreduce.Job: Job job_1541700195986_0073 running in uber mode : false\n",
      "18/11/08 22:14:31 INFO mapreduce.Job:  map 0% reduce 0%\n",
      "18/11/08 22:14:47 INFO mapreduce.Job:  map 51% reduce 0%\n",
      "18/11/08 22:14:53 INFO mapreduce.Job:  map 69% reduce 0%\n",
      "18/11/08 22:14:56 INFO mapreduce.Job:  map 100% reduce 0%\n",
      "18/11/08 22:15:08 INFO mapreduce.Job:  map 100% reduce 25%\n",
      "18/11/08 22:15:10 INFO mapreduce.Job:  map 100% reduce 50%\n",
      "18/11/08 22:15:11 INFO mapreduce.Job:  map 100% reduce 75%\n",
      "18/11/08 22:15:12 INFO mapreduce.Job:  map 100% reduce 100%\n",
      "18/11/08 22:15:12 INFO mapreduce.Job: Job job_1541700195986_0073 completed successfully\n",
      "18/11/08 22:15:12 INFO mapreduce.Job: Counters: 50\n",
      "\tFile System Counters\n",
      "\t\tFILE: Number of bytes read=280966143\n",
      "\t\tFILE: Number of bytes written=422250002\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=5484188\n",
      "\t\tHDFS: Number of read operations=18\n",
      "\t\tHDFS: Number of large read operations=0\n",
      "\t\tHDFS: Number of write operations=8\n",
      "\tJob Counters \n",
      "\t\tKilled reduce tasks=1\n",
      "\t\tLaunched map tasks=2\n",
      "\t\tLaunched reduce tasks=5\n",
      "\t\tData-local map tasks=2\n",
      "\t\tTotal time spent by all maps in occupied slots (ms)=45500\n",
      "\t\tTotal time spent by all reduces in occupied slots (ms)=44151\n",
      "\t\tTotal time spent by all map tasks (ms)=45500\n",
      "\t\tTotal time spent by all reduce tasks (ms)=44151\n",
      "\t\tTotal vcore-milliseconds taken by all map tasks=45500\n",
      "\t\tTotal vcore-milliseconds taken by all reduce tasks=44151\n",
      "\t\tTotal megabyte-milliseconds taken by all map tasks=46592000\n",
      "\t\tTotal megabyte-milliseconds taken by all reduce tasks=45210624\n",
      "\tMap-Reduce Framework\n",
      "\t\tMap input records=4100\n",
      "\t\tMap output records=11472777\n",
      "\t\tMap output bytes=117525361\n",
      "\t\tMap output materialized bytes=140470972\n",
      "\t\tInput split bytes=228\n",
      "\t\tCombine input records=0\n",
      "\t\tCombine output records=0\n",
      "\t\tReduce input groups=369512\n",
      "\t\tReduce shuffle bytes=140470972\n",
      "\t\tReduce input records=11472777\n",
      "\t\tReduce output records=369512\n",
      "\t\tSpilled Records=34418331\n",
      "\t\tShuffled Maps =8\n",
      "\t\tFailed Shuffles=0\n",
      "\t\tMerged Map outputs=8\n",
      "\t\tGC time elapsed (ms)=761\n",
      "\t\tCPU time spent (ms)=76300\n",
      "\t\tPhysical memory (bytes) snapshot=1491148800\n",
      "\t\tVirtual memory (bytes) snapshot=12114399232\n",
      "\t\tTotal committed heap usage (bytes)=1018167296\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=5484188\n",
      "18/11/08 22:15:12 INFO streaming.StreamJob: Output directory: assignment3_1_1541715262509979\n",
      "18/11/08 22:15:14 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032\n",
      "18/11/08 22:15:15 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032\n",
      "18/11/08 22:15:18 INFO mapred.FileInputFormat: Total input files to process : 4\n",
      "18/11/08 22:15:18 INFO mapreduce.JobSubmitter: number of splits:4\n",
      "18/11/08 22:15:18 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1541700195986_0074\n",
      "18/11/08 22:15:18 INFO impl.YarnClientImpl: Submitted application application_1541700195986_0074\n",
      "18/11/08 22:15:18 INFO mapreduce.Job: The url to track the job: http://f3f7668c3905:8088/proxy/application_1541700195986_0074/\n",
      "18/11/08 22:15:18 INFO mapreduce.Job: Running job: job_1541700195986_0074\n",
      "18/11/08 22:15:25 INFO mapreduce.Job: Job job_1541700195986_0074 running in uber mode : false\n",
      "18/11/08 22:15:25 INFO mapreduce.Job:  map 0% reduce 0%\n",
      "18/11/08 22:15:30 INFO mapreduce.Job:  map 100% reduce 0%\n",
      "18/11/08 22:15:37 INFO mapreduce.Job:  map 100% reduce 100%\n",
      "18/11/08 22:15:37 INFO mapreduce.Job: Job job_1541700195986_0074 completed successfully\n",
      "18/11/08 22:15:37 INFO mapreduce.Job: Counters: 49\n",
      "\tFile System Counters\n",
      "\t\tFILE: Number of bytes read=6223239\n",
      "\t\tFILE: Number of bytes written=13146011\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=5484696\n",
      "\t\tHDFS: Number of bytes written=1484298\n",
      "\t\tHDFS: Number of read operations=15\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=4\n",
      "\t\tLaunched reduce tasks=1\n",
      "\t\tData-local map tasks=4\n",
      "\t\tTotal time spent by all maps in occupied slots (ms)=12734\n",
      "\t\tTotal time spent by all reduces in occupied slots (ms)=3839\n",
      "\t\tTotal time spent by all map tasks (ms)=12734\n",
      "\t\tTotal time spent by all reduce tasks (ms)=3839\n",
      "\t\tTotal vcore-milliseconds taken by all map tasks=12734\n",
      "\t\tTotal vcore-milliseconds taken by all reduce tasks=3839\n",
      "\t\tTotal megabyte-milliseconds taken by all map tasks=13039616\n",
      "\t\tTotal megabyte-milliseconds taken by all reduce tasks=3931136\n",
      "\tMap-Reduce Framework\n",
      "\t\tMap input records=369512\n",
      "\t\tMap output records=369512\n",
      "\t\tMap output bytes=5484198\n",
      "\t\tMap output materialized bytes=6223257\n",
      "\t\tInput split bytes=508\n",
      "\t\tCombine input records=0\n",
      "\t\tCombine output records=0\n",
      "\t\tReduce input groups=935\n",
      "\t\tReduce shuffle bytes=6223257\n",
      "\t\tReduce input records=369512\n",
      "\t\tReduce output records=133518\n",
      "\t\tSpilled Records=739024\n",
      "\t\tShuffled Maps =4\n",
      "\t\tFailed Shuffles=0\n",
      "\t\tMerged Map outputs=4\n",
      "\t\tGC time elapsed (ms)=402\n",
      "\t\tCPU time spent (ms)=10270\n",
      "\t\tPhysical memory (bytes) snapshot=1340678144\n",
      "\t\tVirtual memory (bytes) snapshot=10060619776\n",
      "\t\tTotal committed heap usage (bytes)=947388416\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=5484188\n",
      "\tFile Output Format Counters \n",
      "\t\tBytes Written=1484298\n",
      "18/11/08 22:15:37 INFO streaming.StreamJob: Output directory: assignment3_2_1541715262511794\n",
      "cat: Unable to write to output stream.\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "\n",
    "OUT_DIR_1=\"assignment3_1_\"$(date +\"%s%6N\")\n",
    "OUT_DIR_2=\"assignment3_2_\"$(date +\"%s%6N\")\n",
    "NUM_REDUCERS=4\n",
    "\n",
    "# Code for your first job\n",
    "yarn jar /opt/cloudera/parcels/CDH/lib/hadoop-mapreduce/hadoop-streaming.jar \\\n",
    "    -files mapper1.py,reducer1.py \\\n",
    "    -mapper 'python2 mapper1.py' \\\n",
    "    -reducer 'python2 reducer1.py' \\\n",
    "    -numReduceTasks ${NUM_REDUCERS} \\\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",
    "    -D mapreduce.job.output.key.comparator.class=org.apache.hadoop.mapreduce.lib.partition.KeyFieldBasedComparator \\\n",
    "    -D mapreduce.partition.keycomparator.options=\"-k1,3nr\" \\\n",
    "    -files mapper2.py,reducer2.py \\\n",
    "    -mapper 'python2 mapper2.py' \\\n",
    "    -reducer 'python2 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 \"5p;8q\"\n",
    "\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": {
  "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": 2
}