Intro_Hadoop.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/gevargas/bigdata-management/blob/master/Intro_Hadoop.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Introduction to Apache Hadoop with Colab"
      ],
      "metadata": {
        "id": "KxZN8xXrY8ae"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Configuration"
      ],
      "metadata": {
        "id": "xMCuY2jrZF7Z"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "* Hadoop"
      ],
      "metadata": {
        "id": "qofU2ftgiesh"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "HADOOP_VERSION=\"3.4.0\""
      ],
      "metadata": {
        "id": "qZrrtQh_3Isd"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# download hadoop\n",
        "!wget https://downloads.apache.org/hadoop/common/hadoop-{HADOOP_VERSION}/hadoop-{HADOOP_VERSION}.tar.gz"
      ],
      "metadata": {
        "id": "R2_yUY_IZFUw"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "QkCzxzNjRIad"
      },
      "outputs": [],
      "source": [
        "# uncompress\n",
        "!tar -xzf hadoop-{HADOOP_VERSION}.tar.gz\n",
        "!ls"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "* `JAVA_HOME` path"
      ],
      "metadata": {
        "id": "9Ekg-5nghWaH"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# find default colab java path\n",
        "java_home = !readlink -f /usr/bin/java | sed \"s:bin/java::\"   # return a list of size 1\n",
        "java_home = java_home[0]\n",
        "\n",
        "# set JAVA_HOME\n",
        "%env JAVA_HOME={java_home}"
      ],
      "metadata": {
        "id": "gjEr5DAXgOMq"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Extras"
      ],
      "metadata": {
        "id": "uIUcCcKS4wjj"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!rm -r sample_data/           # remove default sample_data folder"
      ],
      "metadata": {
        "id": "BRPb0gZB4yQO"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Runnning Hadoop"
      ],
      "metadata": {
        "id": "_T5ES8DN16I6"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!hadoop-{HADOOP_VERSION}/bin/hadoop --help"
      ],
      "metadata": {
        "id": "gu_p3WigQsQL"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Example 1: Wordcount"
      ],
      "metadata": {
        "id": "uWgklAsC4TR2"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "* Prepare the input files"
      ],
      "metadata": {
        "id": "C58Fx6gt5Y0T"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# copy hadoop configuration xml files to use as input\n",
        "!mkdir input1/\n",
        "!cp hadoop-{HADOOP_VERSION}/etc/hadoop/*.xml  input1/\n",
        "!ls input1"
      ],
      "metadata": {
        "id": "zX8F0z844ee5"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "* Count the number of times `allowed[.]*` appears in the input"
      ],
      "metadata": {
        "id": "37Xoy13s6gtC"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# use on of the mapreduce examples\n",
        "# input:    path containing the text files to use as input\n",
        "# output:   path to store the number of words\n",
        "# grep_exp: regular expresion to use to filter the lines in the input files\n",
        "\n",
        "!hadoop-{HADOOP_VERSION}/bin/hadoop jar hadoop-{HADOOP_VERSION}/share/hadoop/mapreduce/hadoop-mapreduce-examples-{HADOOP_VERSION}.jar \\\n",
        "        grep  \\\n",
        "        input1 \\\n",
        "        output1 \\\n",
        "        'allowed[.]*'\n"
      ],
      "metadata": {
        "id": "eR4uFzayTR_d"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "* see the results"
      ],
      "metadata": {
        "id": "3P4TvHFW9EaE"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!ls output1/"
      ],
      "metadata": {
        "id": "oT69FoII9CO2"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "!cat output1/part-*"
      ],
      "metadata": {
        "id": "Myk3uJEt9K3H"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Example 2: Wordcount using python map & reduce functions"
      ],
      "metadata": {
        "id": "nvH2du327JwE"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "* Collect the dataset"
      ],
      "metadata": {
        "id": "62toGBU_-Wz7"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# 20,000 newsgroup documents partitioned (nearly) evenly across 20 different newsgroups\n",
        "# see http://qwone.com/~jason/20Newsgroups/ for more info\n",
        "\n",
        "!wget http://qwone.com/~jason/20Newsgroups/20news-18828.tar.gz\n",
        "!tar -xzf 20news-18828.tar.gz"
      ],
      "metadata": {
        "id": "mNC_xzqZdcnF"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "* Mapper function"
      ],
      "metadata": {
        "id": "CIkJgRFjlMwm"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "%%writefile mapper.py\n",
        "\n",
        "import sys\n",
        "import io\n",
        "import re\n",
        "import nltk\n",
        "nltk.download('stopwords',quiet=True)\n",
        "\n",
        "from nltk.corpus import stopwords\n",
        "punctuations = '''!()-[]{};:'\"\\,<>./?@#$%^&*_~'''\n",
        "\n",
        "stop_words = set(stopwords.words('english'))\n",
        "input_stream = io.TextIOWrapper(sys.stdin.buffer, encoding='latin1')\n",
        "for line in input_stream:\n",
        "    line = line.strip()\n",
        "    line = re.sub(r'[^\\w\\s]', '',line)\n",
        "    line = line.lower()\n",
        "    for x in line:\n",
        "        if x in punctuations:\n",
        "            line=line.replace(x, \" \")\n",
        "\n",
        "    words=line.split()\n",
        "    for word in words:\n",
        "        if word not in stop_words:\n",
        "            print('%s\\t%s' % (word, 1))\n"
      ],
      "metadata": {
        "id": "jjAQhD7ikkSe"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "* Reducer function"
      ],
      "metadata": {
        "id": "tXOy_onRlUwO"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "%%writefile reducer.py\n",
        "\n",
        "from operator import itemgetter\n",
        "import sys\n",
        "\n",
        "current_word = None\n",
        "current_count = 0\n",
        "word = None\n",
        "\n",
        "# input comes from STDIN\n",
        "for line in sys.stdin:\n",
        "    # remove leading and trailing whitespace\n",
        "    line = line.strip()\n",
        "    line=line.lower()\n",
        "\n",
        "    # parse the input we got from mapper.py\n",
        "    word, count = line.split('\\t', 1)\n",
        "    try:\n",
        "      count = int(count)\n",
        "    except ValueError:\n",
        "      #count was not a number, so silently\n",
        "      #ignore/discard this line\n",
        "      continue\n",
        "\n",
        "    # this IF-switch only works because Hadoop sorts map output\n",
        "    # by key (here: word) before it is passed to the reducer\n",
        "    if current_word == word:\n",
        "        current_count += count\n",
        "    else:\n",
        "        if current_word:\n",
        "            # write result to STDOUT\n",
        "            print ('%s\\t%s' % (current_word, current_count))\n",
        "        current_count = count\n",
        "        current_word = word\n",
        "\n",
        "# do not forget to output the last word if needed!\n",
        "if current_word == word:\n",
        "    print( '%s\\t%s' % (current_word, current_count))\n",
        "\n",
        "\n"
      ],
      "metadata": {
        "id": "AkJlnIBFleLp"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "* Make map/reduce files executables for Hadoop"
      ],
      "metadata": {
        "id": "PAohSjU2m8UE"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!chmod u+rwx mapper.py\n",
        "!chmod u+rwx reducer.py"
      ],
      "metadata": {
        "id": "2_6Uk1F0du8U"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "* Start map reduce using the python files"
      ],
      "metadata": {
        "id": "hK01DxRQq8ir"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!hadoop-{HADOOP_VERSION}/bin/hadoop \\\n",
        "    jar hadoop-{HADOOP_VERSION}/share/hadoop/tools/lib/hadoop-streaming-{HADOOP_VERSION}.jar \\\n",
        "    -input 20news-18828/alt.atheism/49960 \\\n",
        "    -output output2    \\\n",
        "    -file mapper.py   \\\n",
        "    -file reducer.py  \\\n",
        "    -mapper 'python mapper.py' \\\n",
        "    -reducer 'python reducer.py'"
      ],
      "metadata": {
        "id": "Y_pWm6Nod8xF"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "* Verify result"
      ],
      "metadata": {
        "id": "gNB_1n1-pzRK"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!ls output2"
      ],
      "metadata": {
        "id": "2BnWOmjXfy7M"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "!cat output2/part-00000"
      ],
      "metadata": {
        "id": "Whz4O5bFf5DW"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Example 3: mrjob (python library)"
      ],
      "metadata": {
        "id": "2SsMg5Axvigz"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "* Install mrjob"
      ],
      "metadata": {
        "id": "mlpmGPHk6qPN"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install mrjob"
      ],
      "metadata": {
        "id": "Zj6Jcxso6vtk"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "* Evaluate how much time it takes to execute map reduce functions without using hadoop"
      ],
      "metadata": {
        "id": "qv2QOIfMwjFj"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "%%timeit -n 1 -r 3\n",
        "!cat 20news-18828/alt.atheism/49960 | python mapper.py | sort | python reducer.py"
      ],
      "metadata": {
        "id": "2pbVVQXwgYf1"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "* Count frequent words with mrjob"
      ],
      "metadata": {
        "id": "-VzArAaoy3xA"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "%%timeit -n 1 -r 3\n",
        "\n",
        "import sys\n",
        "v = sys.version_info\n",
        "PYTHON_VERSION = str(v.major) + \".\" + str(v.minor)\n",
        "\n",
        "!python /usr/local/lib/python{PYTHON_VERSION}/dist-packages/mrjob/examples/mr_word_freq_count.py 20news-18828/alt.atheism/49960\n"
      ],
      "metadata": {
        "id": "o0wC5MqZ_BHz"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}