Wordcount_in_Spark.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Wordcount in Spark.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "toc_visible": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kPt5q27L5557"
      },
      "source": [
        "# Wordcount in Spark"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "p0-YhEpP_Ds-"
      },
      "source": [
        "### Setup"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Zsj5WYpR9QId"
      },
      "source": [
        "Let's setup Spark on the Colab environment.  Run the cell below!"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "k-qHai2252mI",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "61659f44-87cf-431a-9f57-e4a1642ea281"
      },
      "source": [
        "!pip install pyspark\n",
        "!pip install -U -q PyDrive\n",
        "!apt install openjdk-8-jdk-headless -qq\n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\""
      ],
      "execution_count": 1,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Collecting pyspark\n",
            "  Downloading pyspark-3.1.2.tar.gz (212.4 MB)\n",
            "\u001b[K     |████████████████████████████████| 212.4 MB 67 kB/s \n",
            "\u001b[?25hCollecting py4j==0.10.9\n",
            "  Downloading py4j-0.10.9-py2.py3-none-any.whl (198 kB)\n",
            "\u001b[K     |████████████████████████████████| 198 kB 35.4 MB/s \n",
            "\u001b[?25hBuilding wheels for collected packages: pyspark\n",
            "  Building wheel for pyspark (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for pyspark: filename=pyspark-3.1.2-py2.py3-none-any.whl size=212880768 sha256=31eded5de79abe2117498cb549186935db4396bfbeca160249f369b05f7fd2bf\n",
            "  Stored in directory: /root/.cache/pip/wheels/a5/0a/c1/9561f6fecb759579a7d863dcd846daaa95f598744e71b02c77\n",
            "Successfully built pyspark\n",
            "Installing collected packages: py4j, pyspark\n",
            "Successfully installed py4j-0.10.9 pyspark-3.1.2\n",
            "The following additional packages will be installed:\n",
            "  openjdk-8-jre-headless\n",
            "Suggested packages:\n",
            "  openjdk-8-demo openjdk-8-source libnss-mdns fonts-dejavu-extra\n",
            "  fonts-ipafont-gothic fonts-ipafont-mincho fonts-wqy-microhei\n",
            "  fonts-wqy-zenhei fonts-indic\n",
            "The following NEW packages will be installed:\n",
            "  openjdk-8-jdk-headless openjdk-8-jre-headless\n",
            "0 upgraded, 2 newly installed, 0 to remove and 37 not upgraded.\n",
            "Need to get 36.5 MB of archives.\n",
            "After this operation, 143 MB of additional disk space will be used.\n",
            "Selecting previously unselected package openjdk-8-jre-headless:amd64.\n",
            "(Reading database ... 155047 files and directories currently installed.)\n",
            "Preparing to unpack .../openjdk-8-jre-headless_8u292-b10-0ubuntu1~18.04_amd64.deb ...\n",
            "Unpacking openjdk-8-jre-headless:amd64 (8u292-b10-0ubuntu1~18.04) ...\n",
            "Selecting previously unselected package openjdk-8-jdk-headless:amd64.\n",
            "Preparing to unpack .../openjdk-8-jdk-headless_8u292-b10-0ubuntu1~18.04_amd64.deb ...\n",
            "Unpacking openjdk-8-jdk-headless:amd64 (8u292-b10-0ubuntu1~18.04) ...\n",
            "Setting up openjdk-8-jre-headless:amd64 (8u292-b10-0ubuntu1~18.04) ...\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/orbd to provide /usr/bin/orbd (orbd) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/servertool to provide /usr/bin/servertool (servertool) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/tnameserv to provide /usr/bin/tnameserv (tnameserv) in auto mode\n",
            "Setting up openjdk-8-jdk-headless:amd64 (8u292-b10-0ubuntu1~18.04) ...\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/idlj to provide /usr/bin/idlj (idlj) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/wsimport to provide /usr/bin/wsimport (wsimport) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/jsadebugd to provide /usr/bin/jsadebugd (jsadebugd) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/native2ascii to provide /usr/bin/native2ascii (native2ascii) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/javah to provide /usr/bin/javah (javah) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/hsdb to provide /usr/bin/hsdb (hsdb) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/clhsdb to provide /usr/bin/clhsdb (clhsdb) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/extcheck to provide /usr/bin/extcheck (extcheck) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/schemagen to provide /usr/bin/schemagen (schemagen) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/xjc to provide /usr/bin/xjc (xjc) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/jhat to provide /usr/bin/jhat (jhat) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/wsgen to provide /usr/bin/wsgen (wsgen) in auto mode\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-CJ71AKe91eh"
      },
      "source": [
        "Lets authenticate a Google Drive client to download the file we will be processing in on Spark job."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "5K93ABEy9Zlo"
      },
      "source": [
        "from pydrive.auth import GoogleAuth\n",
        "from pydrive.drive import GoogleDrive\n",
        "from google.colab import auth\n",
        "from oauth2client.client import GoogleCredentials\n",
        "\n",
        "# Authenticate and create the PyDrive client\n",
        "auth.authenticate_user()\n",
        "gauth = GoogleAuth()\n",
        "gauth.credentials = GoogleCredentials.get_application_default()\n",
        "drive = GoogleDrive(gauth)"
      ],
      "execution_count": 2,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "0orRvrc1-545"
      },
      "source": [
        "id='1SE6k_0YukzGd5wK-E4i6mG83nydlfvSa'\n",
        "downloaded = drive.CreateFile({'id': id})\n",
        "downloaded.GetContentFile('pg100.txt')"
      ],
      "execution_count": 3,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "qwtlO4_m_LbQ"
      },
      "source": [
        "By executed the cells above, we can able to see the file *pg100.txt* under the \"Files\" tab on the left panel."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "CRaF2A_j_nC7"
      },
      "source": [
        "### Wordcount"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ebLNUxP0_8x3"
      },
      "source": [
        "After running the setup stage successfully, I am ready to work on the *pg100.txt* file which contains a copy of the complete works of Shakespeare.\n",
        "\n",
        "I am going to write a Spark application which outputs the number of words that start with each letter. This means that for every letter I want to count the total number of (non-unique) words that start with a specific letter. In my implementation, I am **ignoring the letter case**, i.e., consider all words as lower case. Also, I am ignoring all the words **starting** with a non-alphabetic character."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "xu-e7Ph2_ruG"
      },
      "source": [
        "from pyspark.sql import *\n",
        "from pyspark.sql.functions import *\n",
        "from pyspark import SparkContext\n",
        "import pandas as pd\n",
        "\n",
        "# create the Spark Session\n",
        "spark = SparkSession.builder.getOrCreate()\n",
        "\n",
        "# create the Spark Context\n",
        "sc = spark.sparkContext"
      ],
      "execution_count": 4,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "8XRt9sXIpX2M",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "900c2a3a-7ece-4f96-8518-fbc41e5d67f0"
      },
      "source": [
        "!head -5 pg100.txt"
      ],
      "execution_count": 5,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "﻿The Project Gutenberg EBook of The Complete Works of William Shakespeare, by\r\n",
            "William Shakespeare\r\n",
            "\r\n",
            "This eBook is for the use of anyone anywhere at no cost and with\r\n",
            "almost no restrictions whatsoever.  You may copy it, give it away or\r\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "AuAxGFPFB43Y"
      },
      "source": [
        "pg100 = sc.textFile('pg100.txt')"
      ],
      "execution_count": 6,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "A-SSxDs9FdJa"
      },
      "source": [
        "counts = pg100.flatMap(lambda line: line.split(\" \")) \\\n",
        "              .filter(lambda word: len(word) > 0)  \\\n",
        "              .filter(lambda word: ord(word.lower()[0]) in range(ord('a'), ord('z')+1)) \\\n",
        "              .map(lambda word: (word.lower()[0], 1)) \\\n",
        "              .reduceByKey(lambda a, b: a + b)"
      ],
      "execution_count": 7,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "e8XpW34ovQH9",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "66794619-d37c-4217-9969-d1f1b42a821d"
      },
      "source": [
        "counts.collect()"
      ],
      "execution_count": 8,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[('p', 27759),\n",
              " ('g', 20782),\n",
              " ('c', 34567),\n",
              " ('s', 65705),\n",
              " ('b', 45455),\n",
              " ('i', 62167),\n",
              " ('r', 14265),\n",
              " ('y', 25855),\n",
              " ('l', 29569),\n",
              " ('d', 29713),\n",
              " ('j', 3339),\n",
              " ('h', 60563),\n",
              " ('t', 123602),\n",
              " ('e', 18697),\n",
              " ('o', 43494),\n",
              " ('w', 59597),\n",
              " ('f', 36814),\n",
              " ('u', 9170),\n",
              " ('a', 84836),\n",
              " ('n', 26759),\n",
              " ('m', 55676),\n",
              " ('v', 5728),\n",
              " ('k', 9418),\n",
              " ('q', 2377),\n",
              " ('z', 71),\n",
              " ('x', 14)]"
            ]
          },
          "metadata": {},
          "execution_count": 8
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "7jDCs412ZZcF"
      },
      "source": [
        "counts.saveAsTextFile(\"char_count.txt\")"
      ],
      "execution_count": 9,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "6rWykYX4q_v6"
      },
      "source": [
        "sc.stop()"
      ],
      "execution_count": 10,
      "outputs": []
    }
  ]
}