TA_Spark_on_ARC_TALC.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Spark on Arc/Talc.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "toc_visible": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Bs0A_I4Fl-ps"
      },
      "source": [
        "## Initialize Cluster"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "DWgOD_hOwAis",
        "outputId": "3f91ac70-e303-4df8-f8d3-76be95c0c6ea"
      },
      "source": [
        "!python --version"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Python 3.7.0\r\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "WyApJcoaa61_",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "3c23b418-3b48-4eec-81a2-a84ecc1ee59a"
      },
      "source": [
        "!which pip"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "/global/software/jupyterhub-spark/anaconda3/bin/pip\r\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "g1BV-at8FPjM",
        "outputId": "24c66c84-6a18-4046-dba7-fb70361b3ac0"
      },
      "source": [
        "import os\n",
        "import atexit\n",
        "import sys\n",
        "import time\n",
        "\n",
        "import pyspark\n",
        "from pyspark.context import SparkContext\n",
        "from pyspark.sql import SQLContext\n",
        "import findspark\n",
        "from sparkhpc import sparkjob\n",
        "\n",
        "#Exit handler to clean up the Spark cluster if the script exits or crashes\n",
        "def exitHandler(sj,sc):\n",
        "    try:\n",
        "        print('Trapped Exit cleaning up Spark Context')\n",
        "        sc.stop()\n",
        "    except:\n",
        "        pass\n",
        "    try:\n",
        "        print('Trapped Exit cleaning up Spark Job')\n",
        "        sj.stop()\n",
        "    except:\n",
        "        pass\n",
        "\n",
        "findspark.init()\n",
        "\n",
        "#Parameters for the Spark cluster\n",
        "nodes=1\n",
        "tasks_per_node=4 \n",
        "memory_per_task=4096 #4 gig per process, adjust accordingly\n",
        "# Please estimate walltime carefully to keep unused Spark clusters from sitting \n",
        "# idle so that others may use the resources.\n",
        "walltime=\"1:00\" #1 hours\n",
        "#os.environ['SBATCH_PARTITION']='cpu2019' #Set the appropriate ARC partition\n",
        "\n",
        "sj = sparkjob.sparkjob(\n",
        "     ncores=nodes*tasks_per_node,\n",
        "     cores_per_executor=tasks_per_node,\n",
        "     memory_per_core=memory_per_task,\n",
        "     walltime=walltime\n",
        "    )\n",
        "\n",
        "sj.wait_to_start()\n",
        "time.sleep(60)\n",
        "sc = sj.start_spark()\n",
        "\n",
        "#Register the exit handler                                                                                                     \n",
        "atexit.register(exitHandler,sj,sc)\n",
        "\n",
        "#You need this line if you want to use SparkSQL\n",
        "sqlCtx=SQLContext(sc)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "INFO:sparkhpc.sparkjob:Submitted batch job 237\n",
            "\n",
            "INFO:sparkhpc.sparkjob:Submitted cluster 1\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-OPkwC8BIIh1"
      },
      "source": [
        "# Examples"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ouSea4B6IPIQ"
      },
      "source": [
        "## Assignment 1\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "ZGMoXl0cIooh",
        "outputId": "96c02aa1-05a5-4c48-ca67-33782d63ad1c"
      },
      "source": [
        "!pwd"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "/home/ajoy.das/projects/spark\r\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "_cE_l09FImJq"
      },
      "source": [
        "DATA_PATH = '/home/ajoy.das/projects/spark'"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "1sC7NLM-IRlr",
        "outputId": "3ca7bcde-3aeb-40b3-84b1-5205028d7519"
      },
      "source": [
        "import nltk\n",
        "nltk.download('punkt')"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "[nltk_data] Downloading package punkt to /home/ajoy.das/nltk_data...\n",
            "[nltk_data]   Unzipping tokenizers/punkt.zip.\n"
          ]
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "True"
            ]
          },
          "metadata": {},
          "execution_count": 6
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "9D26PZ2DIUO0",
        "outputId": "3531cc18-e1e3-4ebc-bc42-890e07e2f29d"
      },
      "source": [
        "nltk.download('stopwords')\n",
        "from nltk.corpus import stopwords\n",
        "stop_en = stopwords.words('english')\n"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "[nltk_data] Downloading package stopwords to\n",
            "[nltk_data]     /home/ajoy.das/nltk_data...\n",
            "[nltk_data]   Unzipping corpora/stopwords.zip.\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "IVwZyf57IYXp"
      },
      "source": [
        "import pandas as pd\n",
        "from nltk.tokenize import sent_tokenize\n",
        "from nltk.tokenize import word_tokenize\n",
        "from pyspark.sql import Row\n",
        "from bs4 import BeautifulSoup\n",
        " \n",
        "from pyspark.sql.functions import udf\n",
        "from pyspark.ml.feature import Tokenizer\n",
        "from pyspark.ml.feature import StopWordsRemover\n",
        "import pyspark.sql.functions as f"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "J_XmGKoAIckp"
      },
      "source": [
        "sqlContext = sqlCtx\n",
        "spark = sqlCtx\n",
        "\n",
        "@udf(\"String\")\n",
        "def preprocess_udf(body):\n",
        "    body = BeautifulSoup(body)\n",
        " \n",
        "    # noise\n",
        "    urls  =  body.find_all('a')\n",
        "    if len(urls) > 0: body.a.clear()\n",
        " \n",
        "    codes = body.find_all('code')\n",
        "    if len(codes) > 0: body.code.clear()\n",
        " \n",
        "    pres = body.find_all('pre')\n",
        "    if len(pres) > 0: body.pre.clear()\n",
        " \n",
        "        #s = body.find_all('p')\n",
        "        #or p in ps:\n",
        "    text = body.get_text()\n",
        " \n",
        "    words = []\n",
        "    text = text.lower()\n",
        "    sents = nltk.sent_tokenize(text)\n",
        "    for sent in sents:\n",
        "      for word in nltk.word_tokenize(sent):\n",
        "        if word in stop_en: continue\n",
        "        if len(word) < 3: continue\n",
        "        words.append(word)\n",
        "    return \" \".join(words)\n",
        " \n",
        "class SOAnalysis(object):\n",
        "  \n",
        "  def __init__(self, progLang):\n",
        "    pass\n",
        " \n",
        "  def getFile(self):\n",
        "    file_location = DATA_PATH + \"/SO-\"+progLang+\".csv\"\n",
        "    self.df = spark.read.option(\"header\", True).option(\"wholeFile\", True).option(\"escape\", \"\\\"\").option(\"multiLine\", True).csv(file_location)\n",
        "    self.df_prog_body = self.df.select('Body').toPandas()\n",
        "    \n",
        "  def tokenizeAndPreprocess(self, text):\n",
        "    \n",
        "    text = text.lower()\n",
        "    sents = nltk.sent_tokenize(text)\n",
        "    for sent in sents:\n",
        "      for word in nltk.word_tokenize(sent):\n",
        "        if word in stop_en: continue\n",
        "        if len(word) < 3: continue\n",
        "        self.words.append(word)\n",
        " \n",
        "  def doWordCount(self):\n",
        "    bodies = []\n",
        "    self.words = []\n",
        "    isnullBody = pd.isnull(self.df_prog_body[\"Body\"])\n",
        "    noiseTags = [\"a\", \"code\", \"pre\"]\n",
        "    for index, row in self.df_prog_body.iterrows():\n",
        "      if isnullBody[index]: continue\n",
        "      body = row['Body']\n",
        "      body = BeautifulSoup(body)\n",
        " \n",
        "      # noise\n",
        "      urls  =  body.find_all('a')\n",
        "      if len(urls) > 0: body.a.clear()\n",
        "      \n",
        "      codes = body.find_all('code')\n",
        "      if len(codes) > 0: body.code.clear()\n",
        "      \n",
        "      pres = body.find_all('pre')\n",
        "      if len(pres) > 0: body.pre.clear()\n",
        " \n",
        "      #s = body.find_all('p')\n",
        "      #or p in ps:\n",
        "      bodies.append(body.get_text())\n",
        "    for body in bodies:\n",
        "      self.tokenizeAndPreprocess(body)\n",
        "    print(\"total words = \", len(self.words))\n",
        "    rdd1 = sc.parallelize(self.words)\n",
        "    row_rdd = rdd1.map(lambda x: Row(x))\n",
        "    wdf=sqlContext.createDataFrame(row_rdd,['word'])\n",
        "    ddg = wdf.groupBy('word').count()\n",
        "    ddg.orderBy(['count'], ascending=[0]).show(10)\n",
        "  \n",
        "  def getPctAcceptedAnswer(self):\n",
        "    totalAccepted = self.df.where('AcceptedAnswerId is not null').count()\n",
        "    totalQuestions = self.df.count()\n",
        "    if totalQuestions != 0:\n",
        "      pct = totalAccepted * 100.0/ totalQuestions\n",
        "    else:\n",
        "      pct = None\n",
        "    print(\"Percentage of Accepted Answer = %.2f\"%(pct))\n",
        "  \n",
        "  def getQuestionType(self):\n",
        "    dfTitle = self.df.where('Title is not null').select('Title')\n",
        "    qTotal = dfTitle.count()\n",
        "    qWhy = dfTitle.where(\"Title like 'why %'\").count()\n",
        "    qHow = dfTitle.where(\"Title like 'how %'\").count()\n",
        "    qWhat = dfTitle.where(\"Title like 'what %'\").count()\n",
        "    qOther = qTotal - (qWhy + qHow + qWhat)\n",
        "    if qTotal != 0:\n",
        "      pctQWhy = qWhy * 100.0/ qTotal\n",
        "      pctQHow = qHow * 100.0 / qTotal\n",
        "      pctQWhat = qWhat * 100.0 / qTotal\n",
        "      pctQOther = qOther * 100.0 / qTotal\n",
        "      #print(pctQWhat, pctQWhy, pctQHow, pctQOther)\n",
        "    else:\n",
        "      pctQWhy = N, pctQHow, pctQWhat, pctQOther = None, None, None, None\n",
        "    print(\"Pct What = %.2f. Why = %.2f. How = %.2f. Other = %.2f\"%(pctQWhat, pctQWhy, pctQHow, pctQOther))\n",
        "  \n",
        "  def doWordCountSparkOnly(self):\n",
        "    dfBody = self.df.select('Body', preprocess_udf(\"Body\").alias(\"body_cleaned\")).drop(\"Body\")\n",
        "    dfBody.withColumn('word', f.explode(f.split(f.col('body_cleaned'), ' '))).groupBy('word').count().sort('count', ascending=False).show(10)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 204
        },
        "id": "BegxQ2EkIjcw",
        "outputId": "e30f0b4c-02f0-4214-c9b5-7adf4bb73dcf"
      },
      "source": [
        "progLang = 'Python'\n",
        "soAnalysis = SOAnalysis(progLang)\n",
        "soAnalysis.getFile()\n",
        "soAnalysis.df_prog_body.head()\n"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "                                                Body\n",
              "0  <p>I'd like to do some server-side scripting u...\n",
              "1  <p>Can you please tell me how much is <code>(-...\n",
              "2  <p>I am using <code>win32com</code> in python ...\n",
              "3  <p>I'm using pip to install Python libraries o...\n",
              "4  <p>I want to change my Anaconda Prompt User fi..."
            ],
            "text/html": [
              "<div>\n",
              "<style scoped>\n",
              "    .dataframe tbody tr th:only-of-type {\n",
              "        vertical-align: middle;\n",
              "    }\n",
              "\n",
              "    .dataframe tbody tr th {\n",
              "        vertical-align: top;\n",
              "    }\n",
              "\n",
              "    .dataframe thead th {\n",
              "        text-align: right;\n",
              "    }\n",
              "</style>\n",
              "<table border=\"1\" class=\"dataframe\">\n",
              "  <thead>\n",
              "    <tr style=\"text-align: right;\">\n",
              "      <th></th>\n",
              "      <th>Body</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>&lt;p&gt;I'd like to do some server-side scripting u...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>&lt;p&gt;Can you please tell me how much is &lt;code&gt;(-...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>&lt;p&gt;I am using &lt;code&gt;win32com&lt;/code&gt; in python ...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>&lt;p&gt;I'm using pip to install Python libraries o...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>&lt;p&gt;I want to change my Anaconda Prompt User fi...</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ]
          },
          "metadata": {},
          "execution_count": 22
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "Jk9zm8YPJK0j",
        "outputId": "dac58281-7a03-407c-e339-ae07a2c6d7a9"
      },
      "source": [
        "%%time\n",
        "soAnalysis.doWordCount()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "total words =  3506294\n",
            "+------+-----+\n",
            "|  word|count|\n",
            "+------+-----+\n",
            "|  file|33425|\n",
            "|  code|30444|\n",
            "|python|29591|\n",
            "|  like|24039|\n",
            "| using|23305|\n",
            "|   n't|21634|\n",
            "|  line|21368|\n",
            "|   get|20192|\n",
            "|  data|20045|\n",
            "|  want|19666|\n",
            "+------+-----+\n",
            "only showing top 10 rows\n",
            "\n",
            "CPU times: user 3min 29s, sys: 1.28 s, total: 3min 30s\n",
            "Wall time: 3min 36s\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "FOm5Lg3tJQJJ",
        "outputId": "2f67c61c-585d-413e-b000-e1ada5bb02dd"
      },
      "source": [
        "%%time\n",
        "soAnalysis.doWordCountSparkOnly()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+------+-----+\n",
            "|  word|count|\n",
            "+------+-----+\n",
            "|  file|33425|\n",
            "|  code|30444|\n",
            "|python|29591|\n",
            "|  like|24039|\n",
            "| using|23305|\n",
            "|   n't|21634|\n",
            "|  line|21368|\n",
            "|   get|20192|\n",
            "|  data|20045|\n",
            "|  want|19666|\n",
            "+------+-----+\n",
            "only showing top 10 rows\n",
            "\n",
            "CPU times: user 26.6 ms, sys: 10.7 ms, total: 37.3 ms\n",
            "Wall time: 3min 25s\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "6rpdUVaIJT4t",
        "outputId": "0aae4136-f820-484d-f7c6-acc25035bf2b"
      },
      "source": [
        "%%time\n",
        "soAnalysis.getPctAcceptedAnswer()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Percentage of Accepted Answer = 51.10\n",
            "CPU times: user 0 ns, sys: 2.34 ms, total: 2.34 ms\n",
            "Wall time: 1.65 s\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "NaYXek6VJXeJ",
        "outputId": "7a6b77ce-8b9e-43cf-db16-e7fc85330059"
      },
      "source": [
        "%%time\n",
        "soAnalysis.getQuestionType()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Pct What = 0.13. Why = 0.19. How = 1.59. Other = 98.10\n",
            "CPU times: user 4.03 ms, sys: 19 µs, total: 4.05 ms\n",
            "Wall time: 3.14 s\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "P2hwTp6MO0l3"
      },
      "source": [
        "## Assignment 2"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "69BPTCDHPH3d",
        "outputId": "1cf221dc-e77b-4b6c-b63a-763602e19bee"
      },
      "source": [
        "!pip install --user pyspellchecker"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Collecting pyspellchecker\n",
            "\u001b[?25l  Downloading https://files.pythonhosted.org/packages/64/c7/435f49c0ac6bec031d1aba4daf94dc21dc08a9db329692cdb77faac51cea/pyspellchecker-0.6.2-py3-none-any.whl (2.7MB)\n",
            "\u001b[K    100% |████████████████████████████████| 2.7MB 1.9MB/s \n",
            "\u001b[?25hInstalling collected packages: pyspellchecker\n",
            "Successfully installed pyspellchecker-0.6.2\n",
            "\u001b[33mYou are using pip version 10.0.1, however version 21.3.1 is available.\n",
            "You should consider upgrading via the 'pip install --upgrade pip' command.\u001b[0m\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "nCsQzJmUPd4D",
        "outputId": "b4453668-08aa-49fa-fb3b-fbadeef2b0c1"
      },
      "source": [
        "import nltk\n",
        "nltk.download('punkt')\n"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "[nltk_data] Downloading package punkt to /home/ajoy.das/nltk_data...\n",
            "[nltk_data]   Package punkt is already up-to-date!\n"
          ]
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "True"
            ]
          },
          "metadata": {},
          "execution_count": 36
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "XJOsgSP-PZUK",
        "outputId": "6467e358-d983-42f6-e167-ede54edf7e5d"
      },
      "source": [
        "nltk.download('stopwords')\n",
        "from nltk.corpus import stopwords\n",
        "stop_en = stopwords.words('english')\n"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "[nltk_data] Downloading package stopwords to\n",
            "[nltk_data]     /home/ajoy.das/nltk_data...\n",
            "[nltk_data]   Package stopwords is already up-to-date!\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "c2dyzaWnO2xc"
      },
      "source": [
        "import pandas as pd\n",
        "from nltk.tokenize import sent_tokenize\n",
        "from nltk.tokenize import word_tokenize\n",
        "from pyspark.sql import Row\n",
        "from bs4 import BeautifulSoup\n",
        " \n",
        "from pyspark.sql.functions import udf\n",
        "from pyspark.ml.feature import Tokenizer\n",
        "from pyspark.ml.feature import StopWordsRemover\n",
        "import pyspark.sql.functions as f\n",
        "from nltk.stem.porter import *\n",
        "from spellchecker import SpellChecker"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "1P5-5wx2PfLo"
      },
      "source": [
        ""
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "sIfPLs1kPU8_",
        "outputId": "047c5a92-f217-4841-e647-11c8dfdd7741"
      },
      "source": [
        "text = \"this is jsut graet!\"\n",
        "spell = SpellChecker()\n",
        "misspelled = spell.unknown(nltk.word_tokenize(text))\n",
        "misspelled\n"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "{'graet'}"
            ]
          },
          "metadata": {},
          "execution_count": 38
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "4-9478pEPkQJ",
        "outputId": "33f059e4-05c0-41c2-a685-4011996b945e"
      },
      "source": [
        "for word in misspelled:\n",
        "  print(word, \"=>\", spell.correction(word))\n"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "graet => great\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "hg3osOc-PlBH",
        "outputId": "c7433e1b-d96e-4031-c620-09bc3f0cf597"
      },
      "source": [
        "def dot(A,B): \n",
        "    return (sum(a*b for a,b in zip(A,B)))\n",
        "  \n",
        "def getTextDict(a, b):\n",
        "  ab = set(a + b)\n",
        "  ka = {}\n",
        "  kb = {}\n",
        "  for k in ab:\n",
        "    ka[k] = 0\n",
        "    kb[k] =0\n",
        "  for k in a:\n",
        "    ka[k] += 1\n",
        "  for k in b:\n",
        "    kb[k] += 1\n",
        "  \n",
        "  return ka, kb\n",
        " \n",
        "def getCosineSimilarity(ka, kb):\n",
        "  a, b = getTextDict(ka, kb)\n",
        "  a = a.values()\n",
        "  b = b.values()\n",
        "  return dot(a,b) / ( (dot(a,a) **.5) * (dot(b,b) ** .5) )\n",
        " \n",
        "a = ['a', 'a', 'b']\n",
        "b = ['b', 'c', 'c']\n",
        "getCosineSimilarity(a, b)\n"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "0.19999999999999996"
            ]
          },
          "metadata": {},
          "execution_count": 40
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "tBe-wcfzQYlW"
      },
      "source": [
        "spark = sqlCtx"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "taQxln9TPrLH"
      },
      "source": [
        "def preprocess(body):\n",
        "  body = BeautifulSoup(body)\n",
        "  stemmer = PorterStemmer()\n",
        "  # noise\n",
        "  urls  =  body.find_all('a')\n",
        "  if len(urls) > 0: body.a.clear()\n",
        " \n",
        "  codes = body.find_all('code')\n",
        "  if len(codes) > 0: body.code.clear()\n",
        " \n",
        "  pres = body.find_all('pre')\n",
        "  if len(pres) > 0: body.pre.clear()\n",
        "  text = body.get_text()\n",
        " \n",
        "  words = []\n",
        "  text = text.lower()\n",
        "  sents = nltk.sent_tokenize(text)\n",
        "  for sent in sents:\n",
        "    for word in nltk.word_tokenize(sent):\n",
        "      if word in stop_en: continue\n",
        "      if len(word) < 3: continue\n",
        "      words.append(stemmer.stem(word))\n",
        "  return words  \n",
        " \n",
        "class SimilarQuestionRecommender(object):\n",
        "  def __init__(self, progLang):\n",
        "    print(\"loading and preprocessing started ....\")\n",
        "    file_location = DATA_PATH + \"/SO-\"+progLang+\".csv\"\n",
        "    df = spark.read.option(\"header\", True).option(\"wholeFile\", True).option(\"escape\", \"\\\"\").option(\"multiLine\", True).csv(file_location)\n",
        "    pdf = df.select('Id', 'AcceptedAnswerId', 'Score', 'Title', 'Body').toPandas()\n",
        "    pdf['BodyPreprocessed'] = pdf['Body'].map(lambda a: preprocess(a))\n",
        "    pdf['TitlePreprocessed'] = pdf['Title'].map(lambda a: preprocess(a))\n",
        "    pdf['BodyTitle'] = pdf['TitlePreprocessed'] + pdf['BodyPreprocessed'] \n",
        "    self.pdfn = pdf[['Id', 'AcceptedAnswerId', 'Score', 'BodyTitle']]\n",
        "    print(\"loading and preprocessing done ....\")\n",
        "    \n",
        "  def getMaxScoreForQuestionWithoutAcceptedAnswer(self):\n",
        "    maxScore = -99999\n",
        "    isnullAcct = pd.isnull(self.pdfn['AcceptedAnswerId'])\n",
        "    for index, row in self.pdfn.iterrows():\n",
        "      if isnullAcct[index]: \n",
        "        score = int(row['Score'])\n",
        "        if score > maxScore:\n",
        "          maxScore = score\n",
        "    return maxScore\n",
        "  \n",
        "  def getIdsOfQuestionsWithoutAcceptedAnswer(self):\n",
        "    maxScore = self.getMaxScoreForQuestionWithoutAcceptedAnswer()\n",
        "    idsForNonAccepted = []\n",
        "    isnullAcct = pd.isnull(self.pdfn['AcceptedAnswerId'])\n",
        "    for index, row in self.pdfn.iterrows():\n",
        "      if isnullAcct[index]: \n",
        "        score = int(row['Score'])\n",
        "        if score == maxScore:\n",
        "          idsForNonAccepted.append(row[\"Id\"])\n",
        "    return idsForNonAccepted  \n",
        " \n",
        "  def getSimilarQuestionForCandidateWithNotAccepted(self, qid):\n",
        "    sims = dict()\n",
        "    isnullAcct = pd.isnull(self.pdfn['AcceptedAnswerId'])\n",
        "    textN = self.pdfn.loc[self.pdfn['Id'] == str(qid)]['BodyTitle'].values[0]\n",
        "    for index, row in self.pdfn.iterrows():\n",
        "      if isnullAcct[index]: continue\n",
        "      textA = row['BodyTitle']\n",
        "      iid = row['Id']\n",
        "      sim = getCosineSimilarity(textN, textA)\n",
        "      sims[iid] = sim\n",
        "    sims_sorted = sorted(sims.items(), key=lambda x: x[1], reverse=True)\n",
        "    topThree = sims_sorted[0:3]\n",
        "    for item in topThree:\n",
        "      iid = item[0] \n",
        "      sim_score = item[1]\n",
        "      print(\"\\t Question Id = \", iid, \" with Similarity Score = \", sim_score)\n",
        "    #return sims_sorted\n",
        "  \n",
        "  def recommendSimilarQuestion(self):\n",
        "      idsForNonAccepted = self.getIdsOfQuestionsWithoutAcceptedAnswer()\n",
        "      print(\"Total highest scored questions without accepted answers = %d\"%(len(idsForNonAccepted)))\n",
        "      for qid in idsForNonAccepted:\n",
        "        print(\"For \", qid, \": Recommendations are: \")\n",
        "        self.getSimilarQuestionForCandidateWithNotAccepted(qid)\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "6c6yblPIP2qQ",
        "outputId": "87d94a31-f2a8-448f-88fe-5de27f9cabe0"
      },
      "source": [
        "%%time\n",
        "progLang = \"Python\"\n",
        "sqr = SimilarQuestionRecommender(progLang)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "loading and preprocessing started ....\n",
            "loading and preprocessing done ....\n",
            "CPU times: user 5min 54s, sys: 2.77 s, total: 5min 57s\n",
            "Wall time: 6min 20s\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "kVpOX3o0P5dJ",
        "outputId": "2269f12f-0a57-466c-acc4-0caec444baf9"
      },
      "source": [
        "%%time\n",
        "sqr.recommendSimilarQuestion()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Total highest scored questions without accepted answers = 1\n",
            "For  455612 : Recommendations are: \n",
            "\t Question Id =  56433990  with Similarity Score =  0.39605901719066977\n",
            "\t Question Id =  1598579  with Similarity Score =  0.3818813079129867\n",
            "\t Question Id =  8868985  with Similarity Score =  0.380920029682232\n",
            "CPU times: user 20.6 s, sys: 5.79 ms, total: 20.6 s\n",
            "Wall time: 20.8 s\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hL8fPHRpH_0k"
      },
      "source": [
        "## MlLib"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "MDxq71KFMPFk"
      },
      "source": [
        "spark = sqlCtx"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "mctqTQJeNXbN"
      },
      "source": [
        "from pyspark.ml.feature import VectorIndexer, StringIndexer\n",
        "from pyspark.ml.classification import DecisionTreeClassifier\n",
        "from pyspark.ml import Pipeline\n",
        "from pyspark.ml.evaluation import MulticlassClassificationEvaluator"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Av68LAKqmuuG",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "88c4d711-e059-4cff-8bee-43ca8b68ab1f"
      },
      "source": [
        "# Load the data stored in LIBSVM format as a DataFrame.\n",
        "data = spark.read.format(\"libsvm\").load(\"sample_libsvm_data.txt\")\n",
        "\n",
        "# Index labels, adding metadata to the label column.\n",
        "# Fit on whole dataset to include all labels in index.\n",
        "labelIndexer = StringIndexer(inputCol=\"label\", outputCol=\"indexedLabel\").fit(data)\n",
        "# Automatically identify categorical features, and index them.\n",
        "# We specify maxCategories so features with > 4 distinct values are treated as continuous.\n",
        "featureIndexer =\\\n",
        "    VectorIndexer(inputCol=\"features\", outputCol=\"indexedFeatures\", maxCategories=4).fit(data)\n",
        "\n",
        "# Split the data into training and test sets (30% held out for testing)\n",
        "(trainingData, testData) = data.randomSplit([0.7, 0.3])\n",
        "\n",
        "# Train a DecisionTree model.\n",
        "dt = DecisionTreeClassifier(labelCol=\"indexedLabel\", featuresCol=\"indexedFeatures\")\n",
        "\n",
        "# Chain indexers and tree in a Pipeline\n",
        "pipeline = Pipeline(stages=[labelIndexer, featureIndexer, dt])\n",
        "\n",
        "# Train model.  This also runs the indexers.\n",
        "model = pipeline.fit(trainingData)\n",
        "\n",
        "# Make predictions.\n",
        "predictions = model.transform(testData)\n",
        "\n",
        "# Select example rows to display.\n",
        "predictions.select(\"prediction\", \"indexedLabel\", \"features\").show(5)\n",
        "\n",
        "# Select (prediction, true label) and compute test error\n",
        "evaluator = MulticlassClassificationEvaluator(\n",
        "    labelCol=\"indexedLabel\", predictionCol=\"prediction\", metricName=\"accuracy\")\n",
        "accuracy = evaluator.evaluate(predictions)\n",
        "print(\"Test Error = %g \" % (1.0 - accuracy))\n",
        "\n",
        "treeModel = model.stages[2]\n",
        "# summary only\n",
        "print(treeModel)\n",
        "# $example off$"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----------+------------+--------------------+\n",
            "|prediction|indexedLabel|            features|\n",
            "+----------+------------+--------------------+\n",
            "|       1.0|         1.0|(692,[95,96,97,12...|\n",
            "|       1.0|         1.0|(692,[98,99,100,1...|\n",
            "|       1.0|         1.0|(692,[121,122,123...|\n",
            "|       1.0|         1.0|(692,[122,123,124...|\n",
            "|       1.0|         1.0|(692,[123,124,125...|\n",
            "+----------+------------+--------------------+\n",
            "only showing top 5 rows\n",
            "\n",
            "Test Error = 0.0909091 \n",
            "DecisionTreeClassificationModel (uid=DecisionTreeClassifier_7e29319e858c) of depth 2 with 5 nodes\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Or39g-PFWljg"
      },
      "source": [
        ""
      ],
      "execution_count": null,
      "outputs": []
    }
  ]
}