intro-to-hadoop-01.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# <center> Introduction to Hadoop MapReduce </center>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Python Jupyter notebook supports execution of Linux command inside the notebook cells. This is done by adding the **!** to the beginning of the command line. It should be noted that each command begins with a **!** will create a new bash shell and close this cell once the execution is done:\n",
    "- Full path is required\n",
    "- Temporary results and environmental variables will be lost"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We need to initialize Kerberos authentication mechanism"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Interaction with Hadoop Distributed File System is done through `hdfs` and its sub-commands"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./init_hadoop.sh"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!hdfs dfs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## We need to specify the location of the configuration files for our hadoop cluster EVERYTIME we run, otherwise the command will read from the default config location"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!hdfs dfs -ls /"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!hdfs --config ~/hadoop_palmetto/config dfs -ls /"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Challenge\n",
    "\n",
    "Create a directory named **intro-to-hadoop** inside your user directory on HDFS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs --config ~/hadoop_palmetto/config dfs -mkdir -p /user/$USER/intro-to-hadoop"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!hdfs --config ~/hadoop_palmetto/config dfs -ls /user/lngo"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Challenge\n",
    "\n",
    "Upload the **text** directory into the newly created **intro-to-hadoop** directory. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!hdfs --config ~/hadoop_palmetto/config dfs -put"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs --config ~/hadoop_palmetto/config dfs -put text intro-to-hadoop"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Challenge \n",
    "\n",
    "Check the health status of the directories above in HDFS using fsck:\n",
    "```\n",
    "hdfs fsck <path-to-directory> -files -blocks -locations\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs --config ~/hadoop_palmetto/config fsck intro-to-hadoop/text/complete-shakespeare.txt -files -blocks -locations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## MapReduce Programming Paradigm"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**What is “map”?**\n",
    "– A function/procedure that is applied to every individual\n",
    "elements of a collection/list/array/…\n",
    "\n",
    "```\n",
    "int square(x) { return x*x;}\n",
    "map square [1,2,3,4] -> [1,4,9,16]\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**What is “reduce”?**\n",
    "– A function/procedure that performs an operation on a list.\n",
    "This operation will “fold/reduce” this list into a single value\n",
    "(or a smaller subset)\n",
    "\n",
    "```\n",
    "reduce ([1,2,3,4]) using sum -> 10\n",
    "reduce ([1,2,3,4]) using multiply -> 24\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "MapReduce is an old concept in functional programming. It is naturally applicable in HDFS: \n",
    "- `map` tasks are performed on top of individual data blocks (mainly to filter and decrease raw data contents while increase data value\n",
    "- `reduce` tasks are performed on intermediate results from `map` tasks (should now be significantly decreased in size) to calculate the final results. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. The Hello World of Hadoop: Word Count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!mkdir codes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs --config ~/hadoop_palmetto/config dfs -cat intro-to-hadoop/text/complete-shakespeare.txt \\\n",
    "    2>/dev/null | head -n 100"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile codes/wordcountMapper.py\n",
    "#!/software/spackages/linux-centos8-x86_64/gcc-8.3.1/anaconda3-2019.10-v5cuhr6keyz5ryxcwvv2jkzfj2gwrj4a/bin/python                                          \n",
    "import sys                                                                                                \n",
    "for oneLine in sys.stdin:\n",
    "    oneLine = oneLine.strip()\n",
    "    for word in oneLine.split(\" \"):\n",
    "        if word != \"\":\n",
    "            print ('%s\\t%s' % (word, 1)) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs --config ~/hadoop_palmetto/config dfs -cat intro-to-hadoop/text/complete-shakespeare.txt \\\n",
    "    2>/dev/null \\\n",
    "    | head -n 20 \\\n",
    "    | python ./codes/wordcountMapper.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs --config ~/hadoop_palmetto/config dfs -cat intro-to-hadoop/text/complete-shakespeare.txt \\\n",
    "    2>/dev/null \\\n",
    "    | head -n 20 \\\n",
    "    | python ./codes/wordcountMapper.py \\\n",
    "    | sort"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile codes/wordcountReducer.py\n",
    "#!/software/spackages/linux-centos8-x86_64/gcc-8.3.1/anaconda3-2019.10-v5cuhr6keyz5ryxcwvv2jkzfj2gwrj4a/bin/python\n",
    "import sys\n",
    "\n",
    "current_word = None\n",
    "total_word_count = 0\n",
    "\n",
    "for line in sys.stdin:\n",
    "    line = line.strip()\n",
    "    word, count = line.split(\"\\t\", 1)\n",
    "    try:\n",
    "        count = int(count)\n",
    "    except ValueError:\n",
    "        continue\n",
    "    \n",
    "    if current_word == word:\n",
    "        total_word_count += count\n",
    "    else:\n",
    "        if current_word:\n",
    "            print (\"%s\\t%s\" % (current_word, total_word_count))\n",
    "        current_word = word\n",
    "        total_word_count = 1\n",
    "        \n",
    "if current_word == word:\n",
    "    print (\"%s\\t%s\" % (current_word, total_word_count))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs --config ~/hadoop_palmetto/config dfs -cat intro-to-hadoop/text/complete-shakespeare.txt \\\n",
    "    2>/dev/null \\\n",
    "    | head -n 20 \\\n",
    "    | python ./codes/wordcountMapper.py \\\n",
    "    | sort \\\n",
    "    | python ./codes/wordcountReducer.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs --config ~/hadoop_palmetto/config dfs -rm -R intro-to-hadoop/output-wordcount\n",
    "!mapred --config ~/hadoop_palmetto/config streaming \\\n",
    "    -input intro-to-hadoop/text/complete-shakespeare.txt \\\n",
    "    -output intro-to-hadoop/output-wordcount \\\n",
    "    -file ./codes/wordcountMapper.py \\\n",
    "    -mapper wordcountMapper.py \\\n",
    "    -file ./codes/wordcountReducer.py \\\n",
    "    -reducer wordcountReducer.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs --config ~/hadoop_palmetto/config dfs -ls intro-to-hadoop/output-wordcount"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs --config ~/hadoop_palmetto/config dfs -cat intro-to-hadoop/output-wordcount/part-00000 \\\n",
    "    2>/dev/null | head -n 100"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Challenge\n",
    "\n",
    "Modify *wordcountMapper.py* so that punctuations and capitalization are no longer factors in determining unique words"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "%%writefile codes/wordcountEnhancedMapper.py\n",
    "#!/software/spackages/linux-centos8-x86_64/gcc-8.3.1/anaconda3-2019.10-v5cuhr6keyz5ryxcwvv2jkzfj2gwrj4a/bin/python                                          \n",
    "import sys                     \n",
    "import string\n",
    "\n",
    "translator = str.maketrans('', '', string.punctuation)\n",
    "\n",
    "for oneLine in sys.stdin:\n",
    "    oneLine = oneLine.strip()\n",
    "    for word in oneLine.split(\" \"):\n",
    "        if word != \"\":\n",
    "            newWord = word.translate(translator).lower()\n",
    "            print ('%s\\t%s' % (_______, 1)) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -rm -R intro-to-hadoop/output-wordcount-enhanced\n",
    "!ssh dsciutil yarn jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-streaming.jar \\\n",
    "    -input intro-to-hadoop/text/gutenberg-shakespeare.txt \\\n",
    "    -output intro-to-hadoop/output-wordcount \\\n",
    "    -file ____________________________________________________ \\\n",
    "    -mapper _____________________ \\\n",
    "    -file ____________________________________________________ \\\n",
    "    -reducer _____________________ \\"
   ]
  }
 ],
 "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.8.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}