intro-to-hadoop-02.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# <center> Introduction to Hadoop MapReduce </center>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Debugging Hadoop MapReduce Jobs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "** Data: Movie Ratings and Recommendation **\n",
    "\n",
    "An independent movie company is looking to invest in a new movie project. With limited finance, the company wants to \n",
    "analyze the reaction of audiences, particularly toward various movie genres, in order to identify beneficial \n",
    "movie project to focus on. The company relies on data collected from a publicly available recommendation service \n",
    "by [MovieLens](http://dl.acm.org/citation.cfm?id=2827872). This \n",
    "[dataset](http://files.grouplens.org/datasets/movielens/ml-10m-README.html) contains **24404096** ratings and **668953**\n",
    " tag applications across **40110** movies. These data were created by **247753** users between January 09, 1995 and January 29, 2016. This dataset was generated on October 17, 2016. \n",
    "\n",
    "From this dataset, several analyses are possible, include the followings:\n",
    "1.   Find movies which have the highest average ratings over the years and identify the corresponding genre.\n",
    "2.   Find genres which have the highest average ratings over the years.\n",
    "3.   Find users who rate movies most frequently in order to contact them for in-depth marketing analysis.\n",
    "\n",
    "These types of analyses, which are somewhat ambiguous, demand the ability to quickly process large amount of data in \n",
    "elatively short amount of time for decision support purposes. In these situations, the sizes of the data typically \n",
    "make analysis done on a single machine impossible and analysis done using a remote storage system impractical. For \n",
    "remainder of the lessons, we will learn how HDFS provides the basis to store massive amount of data and to enable \n",
    "the programming approach to analyze these data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -ls -h /repository/movielens"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Find movies which have the highest average ratings over the years and report their ratings and genres\n",
    "\n",
    "- Find the average ratings of all movies over the years\n",
    "- Sort the average ratings from highest to lowest\n",
    "- Report the results, augmented by genres"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -ls /repository/movielens"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -cat /repository/movielens/README.txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -cat /repository/movielens/links.csv \\\n",
    "    2>/dev/null | head -n 5"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -cat /repository/movielens/movies.csv \\\n",
    "    2>/dev/null | head -n 5"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -cat /repository/movielens/ratings.csv \\\n",
    "    2>/dev/null | head -n 5"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -cat /repository/movielens/tags.csv \\\n",
    "    2>/dev/null | head -n 5"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Note:\n",
    "\n",
    "To write a MapReduce program, you have to be able to identify the necessary (Key,Value) that can contribute to the final realization of the required results. This is the reducing phase. From this (Key,Value) pair format, you will be able to develop the mapping phase. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%%writefile codes/avgRatingMapper01.py\n",
    "#!/usr/bin/env python\n",
    "\n",
    "import sys\n",
    "\n",
    "for oneMovie in sys.stdin:\n",
    "    oneMovie = oneMovie.strip()\n",
    "    ratingInfo = oneMovie.split(\",\")\n",
    "    movieID = ratingInfo[1]\n",
    "    rating = ratingInfo[2]\n",
    "    print (\"%s\\t%s\" % (movieID, rating)) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -cat /repository/movielens/ratings.csv \\\n",
    "    2>/dev/null | head -n 5 | python ./codes/avgRatingMapper01.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### *Do we really need the headers?*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%%writefile codes/avgRatingMapper02.py\n",
    "#!/usr/bin/env python\n",
    "\n",
    "import sys\n",
    "\n",
    "for oneMovie in sys.stdin:\n",
    "    oneMovie = oneMovie.strip()\n",
    "    ratingInfo = oneMovie.split(\",\")\n",
    "    try:\n",
    "        movieID = ratingInfo[1]\n",
    "        rating = float(ratingInfo[2])\n",
    "        print (\"%s\\t%s\" % (movieID, rating))\n",
    "    except ValueError:\n",
    "        continue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -cat /repository/movielens/ratings.csv \\\n",
    "    2>/dev/null | head -n 5 | python ./codes/avgRatingMapper02.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### *The outcome is correct. Is it useful?*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Getting additional file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!mkdir movielens\n",
    "!hdfs dfs -get /repository/movielens/movies.csv movielens/movies.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%%writefile codes/avgRatingMapper03.py\n",
    "#!/usr/bin/env python\n",
    "\n",
    "import sys\n",
    "import csv\n",
    "\n",
    "movieFile = \"./movielens/movies.csv\"\n",
    "movieList = {}\n",
    "\n",
    "with open(movieFile, mode = 'r') as infile:\n",
    "    reader = csv.reader(infile)\n",
    "    for row in reader:\n",
    "        movieList[row[0]] = {}\n",
    "        movieList[row[0]][\"title\"] = row[1]\n",
    "        movieList[row[0]][\"genre\"] = row[2]\n",
    "\n",
    "for oneMovie in sys.stdin:\n",
    "    oneMovie = oneMovie.strip()\n",
    "    ratingInfo = oneMovie.split(\",\")\n",
    "    try:\n",
    "        movieTitle = movieList[ratingInfo[1]][\"title\"]\n",
    "        movieGenre = movieList[ratingInfo[1]][\"genre\"]\n",
    "        rating = float(ratingInfo[2])\n",
    "        print (\"%s\\t%s\\t%s\" % (movieTitle, rating, movieGenre))\n",
    "    except ValueError:\n",
    "        continue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -cat /repository/movielens/ratings.csv \\\n",
    "    2>/dev/null | head -n 5 | python ./codes/avgRatingMapper03.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### *Test reducer:*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%%writefile codes/avgRatingReducer01.py\n",
    "#!/usr/bin/env python\n",
    "import sys\n",
    "\n",
    "current_movie = None\n",
    "current_rating_sum = 0\n",
    "current_rating_count = 0\n",
    "\n",
    "for line in sys.stdin:\n",
    "    line = line.strip()\n",
    "    movie, rating, genre = line.split(\"\\t\", 2)\n",
    "    try:\n",
    "        rating = float(rating)\n",
    "    except ValueError:\n",
    "        continue\n",
    "\n",
    "    if current_movie == movie:\n",
    "        current_rating_sum += rating\n",
    "        current_rating_count += 1\n",
    "    else:\n",
    "        if current_movie:\n",
    "            rating_average = current_rating_sum / current_rating_count\n",
    "            print (\"%s\\t%s\\t%s\" % (current_movie, rating_average, genre))    \n",
    "        current_movie = movie\n",
    "        current_rating_sum = rating\n",
    "        current_rating_count = 1\n",
    "\n",
    "if current_movie == movie:\n",
    "    rating_average = current_rating_sum / current_rating_count\n",
    "    print (\"%s\\t%s\\t%s\" % (current_movie, rating_average, genre))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -cat /repository/movielens/ratings.csv 2>/dev/null \\\n",
    "    | head -n 5 \\\n",
    "    | python ./codes/avgRatingMapper03.py \\\n",
    "    | sort \\\n",
    "    | python ./codes/avgRatingReducer01.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Non-HDFS correctness test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -cat /repository/movielens/ratings.csv 2>/dev/null \\\n",
    "    | head -n 2000 \\\n",
    "    | python ./codes/avgRatingMapper03.py \\\n",
    "    | grep Matrix"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -cat /repository/movielens/ratings.csv 2>/dev/null \\\n",
    "    | head -n 2000 \\\n",
    "    | python ./codes/avgRatingMapper03.py \\\n",
    "    | grep Matrix \\\n",
    "    | sort \\\n",
    "    | python ./codes/avgRatingReducer01.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Manual calculation check via python\n",
    "(4.0+1.0+5.0)/3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Full execution on HDFS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!yarn jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-streaming.jar \\\n",
    "    -input /repository/movielens/ratings.csv \\\n",
    "    -output intro-to-hadoop/output-movielens-01 \\\n",
    "    -file ./codes/avgRatingMapper03.py \\\n",
    "    -mapper avgRatingMapper03.py \\\n",
    "    -file ./codes/avgRatingReducer01.py \\\n",
    "    -reducer avgRatingReducer01.py \\"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.1 First Error!!!\n",
    "\n",
    "Go back to the first few lines of the previously and look for the INFO line **Submitted application application_xxxx_xxxx**. Running the logs command of yarn with the provided application ID is a straightforward way to access all available log information for that application. The syntax to view yarn log is:\n",
    "\n",
    "```\n",
    "! yarn logs -applicationId APPLICATION_ID\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "# Run the yarn view log command here\n",
    "# Do not run this command in a notebook browser, it will likely crash the browser\n",
    "#!yarn logs -applicationId application_1476193845089_0123"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "However, this information is often massive, as it contains the aggregated logs from all tasks (map and reduce) of the job, which can be in the hundreds. The example below demonstrates this problem by displaying all the possible information of a single-task MapReduce job.\n",
    "In this example, the log of a container has three types of log (LogType): \n",
    "- stderr: Error messages from the actual task execution\n",
    "- stdout: Print out messages if the task includes them\n",
    "- syslog: Logging messages from the Hadoop MapReduce operation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "One approach to reduce the number of possible output is to comment out all non-essential lines (lines containing **INFO**)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!yarn logs -applicationId application_1505269880969_0056 | grep -v INFO"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "Can we refine the information further:\n",
    "- In a MapReduce setting, containers (often) execute the same task.\n",
    "- Can we extract only message listing the Container IDs?\n",
    "\n",
    "~~~\n",
    "!yarn logs -applicationId APPLICATION_ID | grep '^Container:'\n",
    "~~~"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!yarn logs -applicationId application_1505269880969_0056 | grep '^Container:'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Looking at the previous report, we can further identify container information:\n",
    "\n",
    "```\n",
    "Container: container_XXXXXX on  YYYY.palmetto.clemson.edu_ZZZZZ\n",
    "```\n",
    "\n",
    "- Container ID: container_XXXXXX\n",
    "- Address of node where container is placed: YYYY.palmetto.clemson.edu\n",
    "\n",
    "To request yarn to provide a more detailed log at container level, we run:\n",
    "```\n",
    "!yarn logs -applicationId APPLICATION_ID -containerId CONTAINER_ID --nodeAddress NODE_ADDRESS \\\n",
    "    | grep -v INFO\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!yarn logs -applicationId application_1505269880969_0056 \\\n",
    "    -containerId container_e30_1505269880969_0056_01_000012 \\\n",
    "    --nodeAddress dsci035.palmetto.clemson.edu \\\n",
    "    | grep -v INFO"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This error message gives us some insights into the mechanism of Hadoop MapReduce. \n",
    "- Where are the map and reduce python scripts located?\n",
    "- Where would the *movies.csv* file be, if the *-file* flag is used to upload this file?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%%writefile codes/avgRatingMapper04.py\n",
    "#!/usr/bin/env python\n",
    "\n",
    "import sys\n",
    "import csv\n",
    "\n",
    "movieFile = \"./movies.csv\"\n",
    "movieList = {}\n",
    "\n",
    "with open(movieFile, mode = 'r') as infile:\n",
    "    reader = csv.reader(infile)\n",
    "    for row in reader:\n",
    "        movieList[row[0]] = {}\n",
    "        movieList[row[0]][\"title\"] = row[1]\n",
    "        movieList[row[0]][\"genre\"] = row[2]\n",
    "\n",
    "for oneMovie in sys.stdin:\n",
    "    oneMovie = oneMovie.strip()\n",
    "    ratingInfo = oneMovie.split(\",\")\n",
    "    try:\n",
    "        movieTitle = movieList[ratingInfo[1]][\"title\"]\n",
    "        movieGenre = movieList[ratingInfo[1]][\"genre\"]\n",
    "        rating = float(ratingInfo[2])\n",
    "        print (\"%s\\t%s\\t%s\" % (movieTitle, rating, movieGenre))\n",
    "    except ValueError:\n",
    "        continue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!yarn jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-streaming.jar \\\n",
    "    -input /repository/movielens/ratings.csv \\\n",
    "    -output intro-to-hadoop/output-movielens-01 \\\n",
    "    -file ./codes/avgRatingMapper04.py \\\n",
    "    -mapper avgRatingMapper04.py \\\n",
    "    -file ./codes/avgRatingReducer01.py \\\n",
    "    -reducer avgRatingReducer01.py \\\n",
    "    -file ./movielens/movies.csv"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1.2 Second Error!!!\n",
    "\n",
    "- HDFS is read only. Therefore, all output directories must not have existed prior to job submission\n",
    "- This can be resolved either by specifying a new output directory or deleting the existing output directory"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!yarn jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-streaming.jar \\\n",
    "    -input /repository/movielens/ratings.csv \\\n",
    "    -output intro-to-hadoop/output-movielens-02 \\\n",
    "    -file ./codes/avgRatingMapper04.py \\\n",
    "    -mapper avgRatingMapper04.py \\\n",
    "    -file ./codes/avgRatingReducer01.py \\\n",
    "    -reducer avgRatingReducer01.py \\\n",
    "    -file ./movielens/movies.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -ls intro-to-hadoop/output-movielens-02"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -cat intro-to-hadoop/output-movielens-02/part-00000 \\\n",
    "    2>/dev/null | head -n 20"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Challenge:\n",
    "\n",
    "1. Modify *avgRatingReducer02.py* so that only movies with averaged ratings higher than 3.75 are collected\n",
    "2. Further enhance your modification so that not only movies with averaged ratings higher than 3.75 are collected but these movies also need to be rated at least 5000 times. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%%writefile codes/avgRatingMapper04challenge.py\n",
    "#!/usr/bin/env python\n",
    "\n",
    "import sys\n",
    "import csv\n",
    "\n",
    "movieFile = \"./movies.csv\"\n",
    "movieList = {}\n",
    "\n",
    "\n",
    "with open(movieFile, mode = 'r') as infile:\n",
    "    reader = csv.reader(infile)\n",
    "    for row in reader:\n",
    "        movieList[row[0]] = {}\n",
    "        movieList[row[0]][\"title\"] = row[1]\n",
    "        movieList[row[0]][\"genre\"] = row[2]\n",
    "\n",
    "for oneMovie in sys.stdin:\n",
    "    oneMovie = oneMovie.strip()\n",
    "    ratingInfo = oneMovie.split(\",\")\n",
    "    try:\n",
    "        movieTitle = movieList[ratingInfo[1]][\"title\"]\n",
    "        movieGenre = movieList[ratingInfo[1]][\"genre\"]\n",
    "        rating = float(ratingInfo[2])\n",
    "        if _________:\n",
    "            print (\"%s\\t%s\\t%s\" % (movieTitle, rating, movieGenre))\n",
    "    except ValueError:\n",
    "        continue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "!yarn jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-streaming.jar \\\n",
    "    -input /repository/movielens/ratings.csv \\\n",
    "    -output intro-to-hadoop/output-movielens-challenge \\\n",
    "    -file ____________ \\\n",
    "    -mapper ___________ \\\n",
    "    -file ./codes/avgRatingReducer01.py \\\n",
    "    -reducer avgRatingReducer01.py \\\n",
    "    -file ./codes/movielens/movies.csv"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (Anaconda 5.1.0)",
   "language": "python",
   "name": "anaconda3-5.1.0"
  },
  "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.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}