course1-big-data-esentials/c01w03_WordCountTask0.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Hadoop Streaming assignment 0: Word Count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Push 'Open Tool' and get access to your first notebook with the WordCount MapReduce job. Open the notebook, run the job, check that it is successful and push 'Submit assignment'."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing mapper.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile mapper.py\n",
    "\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",
    "    except ValueError as e:\n",
    "        continue\n",
    "    words = re.split(\"\\W*\\s+\\W*\", text, flags=re.UNICODE)\n",
    "    for word in words:\n",
    "        print >> sys.stderr, \"reporter:counter:Wiki stats,Total words,%d\" % 1\n",
    "        print \"%s\\t%d\" % (word.lower(), 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing reducer.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile reducer.py\n",
    "\n",
    "import sys\n",
    "\n",
    "current_key = None\n",
    "word_sum = 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Appending to reducer.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile -a reducer.py\n",
    "\n",
    "for line in sys.stdin:\n",
    "    try:\n",
    "        key, count = line.strip().split('\\t', 1)\n",
    "        count = int(count)\n",
    "    except ValueError as e:\n",
    "        continue\n",
    "    if current_key != key:\n",
    "        if current_key:\n",
    "            print \"%s\\t%d\" % (current_key, word_sum)\n",
    "        word_sum = 0\n",
    "        current_key = key\n",
    "    word_sum += count\n",
    "\n",
    "if current_key:\n",
    "    print \"%s\\t%d\" % (current_key, word_sum)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Found 4 items\r\n",
      "drwxr-xr-x   - hdfs supergroup          0 2017-07-03 23:21 /data/wiki/en_articles\r\n",
      "drwxr-xr-x   - hdfs supergroup          0 2017-07-03 22:37 /data/wiki/en_articles_part\r\n",
      "drwxr-xr-x   - hdfs supergroup          0 2017-08-11 04:12 /data/wiki/en_articles_part1\r\n",
      "-rw-r--r--   2 hdfs supergroup       1914 2017-08-04 16:12 /data/wiki/stop_words_en-xpo6.txt\r\n"
     ]
    }
   ],
   "source": [
    "! hdfs dfs -ls /data/wiki"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0%however\t1\n",
      "0&\\mathrm{if\t1\n",
      "0(8)320-1234\t1\n",
      "0)).(1\t2\n",
      "0,03\t1\n",
      "0,1,...,n\t1\n",
      "0,1,0\t1\n",
      "0,1,\\dots,n\t1\n",
      "0,5\t1\n",
      "0,50\t1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "rm: `wordcount_result_1504600278440756': No such file or directory\n",
      "17/09/05 12:31:23 INFO client.RMProxy: Connecting to ResourceManager at mipt-master.atp-fivt.org/93.175.29.106:8032\n",
      "17/09/05 12:31:23 INFO client.RMProxy: Connecting to ResourceManager at mipt-master.atp-fivt.org/93.175.29.106:8032\n",
      "17/09/05 12:31:24 INFO mapred.FileInputFormat: Total input paths to process : 1\n",
      "17/09/05 12:31:24 INFO mapreduce.JobSubmitter: number of splits:3\n",
      "17/09/05 12:31:24 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1503302131685_0379\n",
      "17/09/05 12:31:25 INFO impl.YarnClientImpl: Submitted application application_1503302131685_0379\n",
      "17/09/05 12:31:25 INFO mapreduce.Job: The url to track the job: http://mipt-master.atp-fivt.org:8088/proxy/application_1503302131685_0379/\n",
      "17/09/05 12:31:25 INFO mapreduce.Job: Running job: job_1503302131685_0379\n",
      "17/09/05 12:31:31 INFO mapreduce.Job: Job job_1503302131685_0379 running in uber mode : false\n",
      "17/09/05 12:31:31 INFO mapreduce.Job:  map 0% reduce 0%\n",
      "17/09/05 12:31:47 INFO mapreduce.Job:  map 33% reduce 0%\n",
      "17/09/05 12:31:49 INFO mapreduce.Job:  map 55% reduce 0%\n",
      "17/09/05 12:31:55 INFO mapreduce.Job:  map 67% reduce 0%\n",
      "17/09/05 12:32:01 INFO mapreduce.Job:  map 78% reduce 0%\n",
      "17/09/05 12:32:11 INFO mapreduce.Job:  map 100% reduce 0%\n",
      "17/09/05 12:32:19 INFO mapreduce.Job:  map 100% reduce 100%\n",
      "17/09/05 12:32:19 INFO mapreduce.Job: Job job_1503302131685_0379 completed successfully\n",
      "17/09/05 12:32:19 INFO mapreduce.Job: Counters: 54\n",
      "\tFile System Counters\n",
      "\t\tFILE: Number of bytes read=4851476\n",
      "\t\tFILE: Number of bytes written=11925190\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=76993447\n",
      "\t\tHDFS: Number of bytes written=5370513\n",
      "\t\tHDFS: Number of read operations=33\n",
      "\t\tHDFS: Number of large read operations=0\n",
      "\t\tHDFS: Number of write operations=16\n",
      "\tJob Counters \n",
      "\t\tLaunched map tasks=3\n",
      "\t\tLaunched reduce tasks=8\n",
      "\t\tRack-local map tasks=3\n",
      "\t\tTotal time spent by all maps in occupied slots (ms)=351040\n",
      "\t\tTotal time spent by all reduces in occupied slots (ms)=231588\n",
      "\t\tTotal time spent by all map tasks (ms)=87760\n",
      "\t\tTotal time spent by all reduce tasks (ms)=38598\n",
      "\t\tTotal vcore-milliseconds taken by all map tasks=87760\n",
      "\t\tTotal vcore-milliseconds taken by all reduce tasks=38598\n",
      "\t\tTotal megabyte-milliseconds taken by all map tasks=179732480\n",
      "\t\tTotal megabyte-milliseconds taken by all reduce tasks=118573056\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=5627293\n",
      "\t\tInput split bytes=390\n",
      "\t\tCombine input records=11937375\n",
      "\t\tCombine output records=575818\n",
      "\t\tReduce input groups=427175\n",
      "\t\tReduce shuffle bytes=5627293\n",
      "\t\tReduce input records=575818\n",
      "\t\tReduce output records=427175\n",
      "\t\tSpilled Records=1151636\n",
      "\t\tShuffled Maps =24\n",
      "\t\tFailed Shuffles=0\n",
      "\t\tMerged Map outputs=24\n",
      "\t\tGC time elapsed (ms)=1453\n",
      "\t\tCPU time spent (ms)=126530\n",
      "\t\tPhysical memory (bytes) snapshot=4855193600\n",
      "\t\tVirtual memory (bytes) snapshot=32990945280\n",
      "\t\tTotal committed heap usage (bytes)=7536115712\n",
      "\t\tPeak Map Physical memory (bytes)=906506240\n",
      "\t\tPeak Map Virtual memory (bytes)=2205544448\n",
      "\t\tPeak Reduce Physical memory (bytes)=281800704\n",
      "\t\tPeak Reduce Virtual memory (bytes)=3315249152\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\tTotal words=11937375\n",
      "\tFile Input Format Counters \n",
      "\t\tBytes Read=76993057\n",
      "\tFile Output Format Counters \n",
      "\t\tBytes Written=5370513\n",
      "17/09/05 12:32:19 INFO streaming.StreamJob: Output directory: wordcount_result_1504600278440756\n",
      "cat: Unable to write to output stream.\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "\n",
    "OUT_DIR=\"wordcount_result_\"$(date +\"%s%6N\")\n",
    "NUM_REDUCERS=8\n",
    "\n",
    "hdfs dfs -rm -r -skipTrash ${OUT_DIR} > /dev/null\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 mapper.py,reducer.py \\\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\n",
    "\n",
    "hdfs dfs -cat ${OUT_DIR}/part-00000 | head"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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": 2
}