Topic_Modelling_with_PySpark_and_Spark_NLP.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Topic Modelling with PySpark and Spark NLP.ipynb",
      "provenance": [],
      "collapsed_sections": [
        "BIszqyp8lzSU",
        "IdZPmaFPtwyq",
        "9hHCLbrpfSqO",
        "Pa4UrSFFvizm",
        "_Oxvz01vfaq6",
        "-OTK4DVHxcR0",
        "ezlCICGwzvQu"
      ],
      "toc_visible": true,
      "authorship_tag": "ABX9TyOozdhjd8hlmlxdkH+C2Shf",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/maobedkova/TopicModelling_PySpark_SparkNLP/blob/master/Topic_Modelling_with_PySpark_and_Spark_NLP.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "rfAJhrgFo5Tv",
        "colab_type": "text"
      },
      "source": [
        "# Topic Modelling with PySpark and Spark NLP"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "1vt8L5e0mss0",
        "colab_type": "text"
      },
      "source": [
        "This is the tutorial for topic modelling using [PySpark](https://spark.apache.org/docs/latest/api/python/index.html) and [Spark NLP](https://www.johnsnowlabs.com/) libraries. This code could be seen as a complement of Topic Modelling with PySpark and Spark NLP blog post on medium. You could refer to this blog post for more elaborated explanation on what topic modelling is, how to use Spark NLP for NLP pipelines and perform topic modelling with PySpark."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "CZ9ZF9pQqW2X",
        "colab_type": "text"
      },
      "source": [
        "The code shows how to install PySpark and Spark NLP libraries and download a Kaggle dataset to Google Collaboratory. It also illustrates how to build the NLP pipeline with Spark NLP and train a topic model with PySpark."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "BIszqyp8lzSU",
        "colab_type": "text"
      },
      "source": [
        "## Installation"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "CMxW9ewNmaPX",
        "colab_type": "text"
      },
      "source": [
        "The initial task is to install PySpark and Spark NLP libraries to our Google Collaboratory and set the data up. We can proceed with the following installation. We experiment with the latest versions of PySpark (2.4.5) and Spark NLP (2.4.5) to date. We also install `nltk` package because we will need it in one of the steps of our pipeline."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "yPmFVedSA68v",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import os\n",
        "\n",
        "# Install java\n",
        "! apt-get install -y openjdk-8-jdk-headless -qq > /dev/null\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "os.environ[\"PATH\"] = os.environ[\"JAVA_HOME\"] + \"/bin:\" + os.environ[\"PATH\"]\n",
        "! java -version\n",
        "\n",
        "# Install pyspark\n",
        "! pip install --ignore-installed pyspark==2.4.5\n",
        "\n",
        "# Install Spark NLP\n",
        "! pip install --ignore-installed spark-nlp==2.4.5\n",
        "\n",
        "# Install nltk\n",
        "! pip install nltk"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "KfgD9jhRrFEF",
        "colab_type": "text"
      },
      "source": [
        "We are going to use data from [Kaggle](https://www.kaggle.com//). I have chosen [Amazon Musical Instruments Review](https://www.kaggle.com/) dataset for our topic modelling analysis. The data is not that big to see the benefit of using Spark but it is nice for tutorial purposes. Unfortunately, you need to be registered to download the data. Sign in to Kaggle, go to `My Account` and create `New API Token`. This will download `kaggle.json` to your computer."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0qRFqVCfo_HY",
        "colab_type": "text"
      },
      "source": [
        "To use Google Drive where we will put our Kaggle data in Google Collaboratory we need to mount our Google Drive. Follow the instructions after you run the next cell."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "sga628Xqypd6",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from google.colab import drive\n",
        "\n",
        "drive.mount('/content/gdrive')"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_6WvJGgvsTze",
        "colab_type": "text"
      },
      "source": [
        "We also need to upload `kaggle.json`. Follow the instructions after you run the next cell."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "jvNOT0ZtqimJ",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from google.colab import files\n",
        "\n",
        "files.upload()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "q9hibzBTstKT",
        "colab_type": "text"
      },
      "source": [
        "Further, we download the chosen Kaggle dataset to our Google Drive."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "EpEqGfu5qvhf",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!pip install -q kaggle\n",
        "!mkdir -p ~/.kaggle\n",
        "!cp kaggle.json ~/.kaggle/\n",
        "!ls ~/.kaggle\n",
        "!chmod 600 /root/.kaggle/kaggle.json\n",
        "!kaggle datasets download -d eswarchandt/amazon-music-reviews -p '/content/gdrive/My Drive/Colab Notebooks/data/'\n",
        "!unzip '/content/gdrive/My Drive/Colab Notebooks/data/amazon-music-reviews.zip' -d '/content/gdrive/My Drive/Colab Notebooks/data/'"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "5D07dyhss2Hk",
        "colab_type": "text"
      },
      "source": [
        "The data is ready to use. Let's start the Spark session through Spark NLP. If you need a special Spark session setting, you can start Spark session with `SparkSession.builder` from PySpark."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "_E4acZCg8Ku8",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import sparknlp\n",
        "\n",
        "spark = sparknlp.start()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "IdZPmaFPtwyq",
        "colab_type": "text"
      },
      "source": [
        "## Data"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Twlcj7xFt318",
        "colab_type": "text"
      },
      "source": [
        "Here we start our topic modelling tutorial. First, we access the downloaded data and read it into Spark dataframe."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "19uyI_8dAd3j",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.sql import functions as F\n",
        "\n",
        "data_path = '/content/gdrive/My Drive/Colab Notebooks/data/Musical_instruments_reviews.csv'\n",
        "data = spark.read.csv(data_path, header=True)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "BohpGAxPuezM",
        "colab_type": "text"
      },
      "source": [
        "Let's checkout what kind of information is stored in our data."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "x_4_zyfNBlzB",
        "colab_type": "code",
        "outputId": "26f0b424-91c5-4e79-b3b8-99e30e1ac1ce",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 168
        }
      },
      "source": [
        "data.columns"
      ],
      "execution_count": 8,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "['reviewerID',\n",
              " 'asin',\n",
              " 'reviewerName',\n",
              " 'helpful',\n",
              " 'reviewText',\n",
              " 'overall',\n",
              " 'summary',\n",
              " 'unixReviewTime',\n",
              " 'reviewTime']"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 8
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "o2yf6V87uk9F",
        "colab_type": "text"
      },
      "source": [
        "For topic modelling, we need only textual data, thus, we create a new dataframe only with the column of interest."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "h71u8qFR7g40",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "text_col = 'reviewText'\n",
        "review_text = data.select(text_col).filter(F.col(text_col).isNotNull())"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "quLpRWAEu2Gd",
        "colab_type": "text"
      },
      "source": [
        "The data that we will use further for the analysis looks as follows:"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "cID1bvIuyWgn",
        "colab_type": "code",
        "outputId": "fb879f7d-0c8d-4c35-dc38-761c11bca5c9",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 185
        }
      },
      "source": [
        "review_text.limit(5).show(truncate=90)"
      ],
      "execution_count": 10,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+------------------------------------------------------------------------------------------+\n",
            "|                                                                                reviewText|\n",
            "+------------------------------------------------------------------------------------------+\n",
            "|                                                                          that's just like|\n",
            "|The product does exactly as it should and is quite affordable.I did not realized it was...|\n",
            "|The primary job of this device is to block the breath that would otherwise produce a po...|\n",
            "|Nice windscreen protects my MXL mic and prevents pops. Only thing is that the gooseneck...|\n",
            "|This pop filter is great. It looks and performs like a studio filter. If you're recordi...|\n",
            "+------------------------------------------------------------------------------------------+\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "9hHCLbrpfSqO",
        "colab_type": "text"
      },
      "source": [
        "## Spark NLP pipeline"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "i36XwEh8vL85",
        "colab_type": "text"
      },
      "source": [
        "Here we start our NLP pipeline for the task of topic modelling."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Pa4UrSFFvizm",
        "colab_type": "text"
      },
      "source": [
        "### Basic NLP pipeline"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ELN2q7q-vppo",
        "colab_type": "text"
      },
      "source": [
        "Let's start with basic NLP pipeline that clears the data and gets lemmatized unigrams. To understand how you can use Spark NLP annotators (estimators and transformers) for NLP pipeline, you can refer to [Spark NLP documentation](https://nlp.johnsnowlabs.com/) or a corresponding blog post on Topic Modelling with PySpark and Spark NLP."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "4PjxUquwv83c",
        "colab_type": "text"
      },
      "source": [
        "We will start with [**DocumentAssembler**](https://nlp.johnsnowlabs.com/docs/en/transformers#documentassembler-getting-data-in) that converts data into Spark NLP annotation format that can be used by Spark NLP annotators."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "XwOdzQ_PAJi0",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from sparknlp.base import DocumentAssembler\n",
        "\n",
        "documentAssembler = DocumentAssembler() \\\n",
        "     .setInputCol(text_col) \\\n",
        "     .setOutputCol('document')"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "facUm7iMwUKZ",
        "colab_type": "text"
      },
      "source": [
        "Next step is to tokenize data with [**Tokenizer**](https://nlp.johnsnowlabs.com/docs/en/annotators#tokenizer)."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "IGOLmEzJDCYW",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from sparknlp.annotator import Tokenizer\n",
        "\n",
        "tokenizer = Tokenizer() \\\n",
        "     .setInputCols(['document']) \\\n",
        "     .setOutputCol('tokenized')"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "uA98q-HCxVjb",
        "colab_type": "text"
      },
      "source": [
        "Further, we clean our data and lowercase it with [**Normalizer**](https://nlp.johnsnowlabs.com/docs/en/annotators#normalizer)."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "dTuQv1VXD8xu",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from sparknlp.annotator import Normalizer\n",
        "\n",
        "normalizer = Normalizer() \\\n",
        "     .setInputCols(['tokenized']) \\\n",
        "     .setOutputCol('normalized') \\\n",
        "     .setLowercase(True)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "80FaYj0axg1r",
        "colab_type": "text"
      },
      "source": [
        "We are going to lemmatize our text with pretrained lemming model provided by Spark NLP. We can access this model with [**LemmatizerModel**](https://nlp.johnsnowlabs.com/docs/en/models#english---models)."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "a1lvJInOEEAy",
        "colab_type": "code",
        "outputId": "1c707936-8490-4154-a028-5aee58ce34a8",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 67
        }
      },
      "source": [
        "from sparknlp.annotator import LemmatizerModel\n",
        "\n",
        "lemmatizer = LemmatizerModel.pretrained() \\\n",
        "     .setInputCols(['normalized']) \\\n",
        "     .setOutputCol('lemmatized')"
      ],
      "execution_count": 14,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "lemma_antbnc download started this may take some time.\n",
            "Approximate size to download 907.6 KB\n",
            "[OK!]\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0-fkFtA9yNEc",
        "colab_type": "text"
      },
      "source": [
        "Spark NLP doesn't provide stop word list, hence, we will use `nltk` package to download stop words for English."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Um9iOifSft2t",
        "colab_type": "code",
        "outputId": "6ea110e1-2f86-4ff2-a715-d22822d7548e",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 50
        }
      },
      "source": [
        "import nltk\n",
        "nltk.download('stopwords')\n",
        "\n",
        "from nltk.corpus import stopwords\n",
        "\n",
        "eng_stopwords = stopwords.words('english')"
      ],
      "execution_count": 15,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "[nltk_data] Downloading package stopwords to /root/nltk_data...\n",
            "[nltk_data]   Unzipping corpora/stopwords.zip.\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "zKDZh-kkybKt",
        "colab_type": "text"
      },
      "source": [
        "The downloaded list of stop words we will input into [**StopWordsCleaner**](https://nlp.johnsnowlabs.com/docs/en/annotators#stopwordscleaner) that will remove all such words from our lemmatized text."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "RJOGb5argbrb",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from sparknlp.annotator import StopWordsCleaner\n",
        "\n",
        "stopwords_cleaner = StopWordsCleaner() \\\n",
        "     .setInputCols(['lemmatized']) \\\n",
        "     .setOutputCol('unigrams') \\\n",
        "     .setStopWords(eng_stopwords)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tW87_1g2yr3-",
        "colab_type": "text"
      },
      "source": [
        "In addition to unigrams, it is good to use n-grams for topic modelling as well since they help to better refine topics. We can get n-grams with [**NGramGenerator**](https://nlp.johnsnowlabs.com/docs/en/annotators#ngramgenerator) in Spark NLP."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "OsftJKunZO84",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from sparknlp.annotator import NGramGenerator\n",
        "\n",
        "ngrammer = NGramGenerator() \\\n",
        "    .setInputCols(['lemmatized']) \\\n",
        "    .setOutputCol('ngrams') \\\n",
        "    .setN(3) \\\n",
        "    .setEnableCumulative(True) \\\n",
        "    .setDelimiter('_')"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ymL8JUPrzV0R",
        "colab_type": "text"
      },
      "source": [
        "We already have our basic NLP pipeline for topic modelling with all necessary steps. However, let's use POS tagger in order to improve our processed data for topic modelling even more with POS tagged data later. For this, we are going to use pretrained POS tagging model provided by Spark NLP. We can access the model with [**PerceptronModel**](https://nlp.johnsnowlabs.com/docs/en/annotators#postagger)."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ilOwKkC20Xe5",
        "colab_type": "code",
        "outputId": "ac663115-1707-4912-f815-a80e109d113a",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 67
        }
      },
      "source": [
        "from sparknlp.annotator import PerceptronModel\n",
        "\n",
        "pos_tagger = PerceptronModel.pretrained('pos_anc') \\\n",
        "    .setInputCols(['document', 'lemmatized']) \\\n",
        "    .setOutputCol('pos')"
      ],
      "execution_count": 18,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "pos_anc download started this may take some time.\n",
            "Approximate size to download 4.3 MB\n",
            "[OK!]\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Hl6Xs_zE0Fc8",
        "colab_type": "text"
      },
      "source": [
        "Now we have everything in Spark NLP annotation format. To be able to process the data further, we need to tranform data with [**Finisher**](https://nlp.johnsnowlabs.com/docs/en/transformers#finisher)."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "g3eYHx-KrDxU",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from sparknlp.base import Finisher\n",
        "\n",
        "finisher = Finisher() \\\n",
        "     .setInputCols(['unigrams', 'ngrams', 'pos']) \\"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "DP-kJKHQ0T8e",
        "colab_type": "text"
      },
      "source": [
        "Now we are ready to input everything into a pipeline. **Pipeline** functionality is accessible with PySpark."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Oo4ixwfsrMPh",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml import Pipeline\n",
        "\n",
        "pipeline = Pipeline() \\\n",
        "     .setStages([documentAssembler,                  \n",
        "                 tokenizer,\n",
        "                 normalizer,                  \n",
        "                 lemmatizer,                  \n",
        "                 stopwords_cleaner, \n",
        "                 pos_tagger,\n",
        "                 ngrammer,  \n",
        "                 finisher])"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0yXiHvEE0gYx",
        "colab_type": "text"
      },
      "source": [
        "First, we will fit all our estimators and then, transform the data with trained models and transformers."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "pCkceTwhrOf0",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "processed_review = pipeline.fit(review_text).transform(review_text)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "t7ovKGdr0p6J",
        "colab_type": "text"
      },
      "source": [
        "Let's look at the data we finally get."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "D-vuMJZD9Lwc",
        "colab_type": "code",
        "outputId": "a605b4ee-fbf9-4dd6-93fe-d1489adcbca7",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 185
        }
      },
      "source": [
        "processed_review.limit(5).show()"
      ],
      "execution_count": 22,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+--------------------+--------------------+--------------------+--------------------+\n",
            "|          reviewText|   finished_unigrams|     finished_ngrams|        finished_pos|\n",
            "+--------------------+--------------------+--------------------+--------------------+\n",
            "|    that's just like|              [like]|[that, just, like...|        [IN, RB, IN]|\n",
            "|The product does ...|[product, exactly...|[the, product, do...|[DT, NN, VBP, RB,...|\n",
            "|The primary job o...|[primary, job, de...|[the, primary, jo...|[DT, JJ, NN, IN, ...|\n",
            "|Nice windscreen p...|[nice, windscreen...|[nice, windscreen...|[JJ, NN, NN, NNP,...|\n",
            "|This pop filter i...|[pop, filter, gre...|[this, pop, filte...|[DT, NN, NN, VB, ...|\n",
            "+--------------------+--------------------+--------------------+--------------------+\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_Oxvz01vfaq6",
        "colab_type": "text"
      },
      "source": [
        "### Extended NLP pipeline"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tNShG0il00g6",
        "colab_type": "text"
      },
      "source": [
        "Up to now, we have our data in a form of unigrams that are lemmatized, with no stop words in there. I think it is a good idea to incorporate n-grams into our NLP pipeline. We obtained n-grams as one step of our pipeline but now n-grams are messy and have a lot of questionable combinations in there. To tackle this problem, let's filter out strange combinations of words in n-grams based on their POS tags. We can imagine a list of viable combinations like ADJ + NOUN so let's restrict our POS combinations in n-grams to this list. Plus, we can also exclude some POS tags from our unigrams to ensure that we don't use functional words for topic modelling (they can be partially covered by stop words but probably not fully).\n",
        "\n",
        "Doing this POS-based filtering will significantly reduce the vocabulary size for topic modelling which will speed up the whole processing."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-FnnKmIf2Ssc",
        "colab_type": "text"
      },
      "source": [
        "Let's start this processing. First, we need join all our POS tags obtained previously."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "WbPpOs8ndGE9",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.sql import types as T\n",
        "\n",
        "udf_join_arr = F.udf(lambda x: ' '.join(x), T.StringType())\n",
        "processed_review  = processed_review.withColumn('finished_pos', udf_join_arr(F.col('finished_pos')))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "IzSryqMc2an7",
        "colab_type": "text"
      },
      "source": [
        "Then we start another Spark NLP pipeline in order to get POS tag n-grams that correspond to word n-grams. We start with convertation into Spark NLP annotation format."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "RrKEpmfoba6M",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "pos_documentAssembler = DocumentAssembler() \\\n",
        "     .setInputCol('finished_pos') \\\n",
        "     .setOutputCol('pos_document')"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "smZ6-1dz2uf3",
        "colab_type": "text"
      },
      "source": [
        "Then, we tokenize our POS tags."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "pyZdZ7S8aGZ9",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "pos_tokenizer = Tokenizer() \\\n",
        "     .setInputCols(['pos_document']) \\\n",
        "     .setOutputCol('pos')"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Z8kJmdm52w-M",
        "colab_type": "text"
      },
      "source": [
        "And generate n-grams from them in the same way we did that for words."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "9QnL0zwrW_w-",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "pos_ngrammer = NGramGenerator() \\\n",
        "    .setInputCols(['pos']) \\\n",
        "    .setOutputCol('pos_ngrams') \\\n",
        "    .setN(3) \\\n",
        "    .setEnableCumulative(True) \\\n",
        "    .setDelimiter('_')"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "vG3UTh0N23Yf",
        "colab_type": "text"
      },
      "source": [
        "Lastly, we are ready to get POS tags ngrams with **Finisher**."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "CiUGFlMHZqIZ",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "pos_finisher = Finisher() \\\n",
        "     .setInputCols(['pos', 'pos_ngrams']) \\"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "oSQ1XeNZ2--r",
        "colab_type": "text"
      },
      "source": [
        "We create this new Spark NLP pipeline..."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "_-AufeTzbYwM",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "pos_pipeline = Pipeline() \\\n",
        "     .setStages([pos_documentAssembler,                  \n",
        "                 pos_tokenizer,\n",
        "                 pos_ngrammer,  \n",
        "                 pos_finisher])"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "bxQJcpiA3EBg",
        "colab_type": "text"
      },
      "source": [
        "... and again fit it and transform the data."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "0kfpx2azbmJI",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "processed_review = pos_pipeline.fit(processed_review).transform(processed_review)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cUbDHzkZ3JzH",
        "colab_type": "text"
      },
      "source": [
        "Let's look what kind of data we have to operate with."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "x0dBA85zf3Sd",
        "colab_type": "code",
        "outputId": "0b6b9286-32b7-43bc-83eb-aa7d46588f93",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 101
        }
      },
      "source": [
        "processed_review.columns"
      ],
      "execution_count": 30,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "['reviewText',\n",
              " 'finished_unigrams',\n",
              " 'finished_ngrams',\n",
              " 'finished_pos',\n",
              " 'finished_pos_ngrams']"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 30
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "U5VXa6ir3Q97",
        "colab_type": "text"
      },
      "source": [
        "And these are our word n-grams with their corresponding pos n-grams."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "dPRHCY32eKAH",
        "colab_type": "code",
        "outputId": "ed75f929-a561-4dc4-c480-7e90e1a0bac9",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 185
        }
      },
      "source": [
        "processed_review.select('finished_ngrams', 'finished_pos_ngrams').limit(5).show()"
      ],
      "execution_count": 31,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+--------------------+--------------------+\n",
            "|     finished_ngrams| finished_pos_ngrams|\n",
            "+--------------------+--------------------+\n",
            "|[that, just, like...|[IN, RB, IN, IN_R...|\n",
            "|[the, product, do...|[DT, NN, VBP, RB,...|\n",
            "|[the, primary, jo...|[DT, JJ, NN, IN, ...|\n",
            "|[nice, windscreen...|[JJ, NN, NN, NNP,...|\n",
            "|[this, pop, filte...|[DT, NN, NN, VB, ...|\n",
            "+--------------------+--------------------+\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XCXs_3yNfXbG",
        "colab_type": "text"
      },
      "source": [
        "Now we are ready to filter out not useful for topic modelling analysis POS tags from our data. Let's create the function that does it for unigrams first. We create the custom Python function and then transform it to PySpark UDF to be used on Spark dataframe."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "YhhKK97NfEbd",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "def filter_pos(words, pos_tags):\n",
        "    return [word for word, pos in zip(words, pos_tags) \n",
        "            if pos in ['JJ', 'NN', 'NNS', 'VB', 'VBP']]\n",
        "\n",
        "udf_filter_pos = F.udf(filter_pos, T.ArrayType(T.StringType()))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "k_hy63My3rrV",
        "colab_type": "text"
      },
      "source": [
        "Then, we apply this function on columns with unigrams and their POS tags to get filtered unigrams in a separate dataframe column."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "rEx5CHyqfWeG",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "processed_review = processed_review.withColumn('filtered_unigrams',\n",
        "                                               udf_filter_pos(F.col('finished_unigrams'), \n",
        "                                                              F.col('finished_pos')))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Yh1XVgPL4K6G",
        "colab_type": "text"
      },
      "source": [
        "That is how our filtered unigrams look like."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "L22iUDgWgMVd",
        "colab_type": "code",
        "outputId": "a39d6f9d-fb38-4b96-c97c-7e8d6ce35e6e",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 185
        }
      },
      "source": [
        "processed_review.select('filtered_unigrams').limit(5).show(truncate=90)"
      ],
      "execution_count": 34,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+------------------------------------------------------------------------------------------+\n",
            "|                                                                         filtered_unigrams|\n",
            "+------------------------------------------------------------------------------------------+\n",
            "|                                                                                        []|\n",
            "|[exactly, quite, even, expectedas, add, one, carry, small, hint, grape, buy, put, next,...|\n",
            "|[job, device, would, otherwise, pop, allow, reduction, high, frequency, cloth, block, l...|\n",
            "|[nice, windscreen, protect, mic, prevent, thing, gooseneck, able, hold, require, carefu...|\n",
            "|               [filter, great, look, perform, studio, youre, eliminate, pop, record, sing]|\n",
            "+------------------------------------------------------------------------------------------+\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "GEDG4jfD4Rux",
        "colab_type": "text"
      },
      "source": [
        "It is time to filter out improper POS combinations of n-grams. We create the custom function in the same manner as before. Since we deal with bi- and trigrams, we need to restrict tags for both."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "yJd-nLnxgd1g",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "def filter_pos_combs(words, pos_tags):\n",
        "    return [word for word, pos in zip(words, pos_tags) \n",
        "            if (len(pos.split('_')) == 2 and \\\n",
        "                pos.split('_')[0] in ['JJ', 'NN', 'NNS', 'VB', 'VBP'] and \\\n",
        "                 pos.split('_')[1] in ['JJ', 'NN', 'NNS']) \\\n",
        "            or (len(pos.split('_')) == 3 and \\\n",
        "                pos.split('_')[0] in ['JJ', 'NN', 'NNS', 'VB', 'VBP'] and \\\n",
        "                 pos.split('_')[1] in ['JJ', 'NN', 'NNS', 'VB', 'VBP'] and \\\n",
        "                  pos.split('_')[2] in ['NN', 'NNS'])]\n",
        "    \n",
        "udf_filter_pos_combs = F.udf(filter_pos_combs, T.ArrayType(T.StringType()))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "4Zjj8b9v425f",
        "colab_type": "text"
      },
      "source": [
        "And we call the function on word and POS n-grams."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "QH1gNrz6giPU",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "processed_review = processed_review.withColumn('filtered_ngrams',\n",
        "                                               udf_filter_pos_combs(F.col('finished_ngrams'),\n",
        "                                                                    F.col('finished_pos_ngrams')))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_kA7epRY5HYy",
        "colab_type": "text"
      },
      "source": [
        "Below is what we get after filtering for n-grams."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "gcXY0eMChtTw",
        "colab_type": "code",
        "outputId": "5a9626e9-19d5-4685-8dc5-47d5e372599d",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 185
        }
      },
      "source": [
        "processed_review.select('filtered_ngrams').limit(5).show(truncate=90)"
      ],
      "execution_count": 37,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+------------------------------------------------------------------------------------------+\n",
            "|                                                                           filtered_ngrams|\n",
            "+------------------------------------------------------------------------------------------+\n",
            "|                                                                                        []|\n",
            "|[be_double, double_screen, add_bonus, small_hint, old_grape, grape_candy, cannot_stop, ...|\n",
            "|[primary_job, pop_sound, noticeable_reduction, high_frequency, double_cloth, cloth_filt...|\n",
            "|[nice_windscreen, windscreen_protect, mxl_mic, prevent_pop, require_careful, careful_po...|\n",
            "|             [pop_filter, be_great, studio_filter, youre_record, record_vocal, get_record]|\n",
            "+------------------------------------------------------------------------------------------+\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ciCceJZH5NmN",
        "colab_type": "text"
      },
      "source": [
        "Now we have unigrams and n-grams stored in different columns in the dataframe. Let's combine them together."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GVNl50pI7U1S",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.sql.functions import concat\n",
        "\n",
        "processed_review = processed_review.withColumn('final', \n",
        "                                               concat(F.col('filtered_unigrams'), \n",
        "                                                      F.col('filtered_ngrams')))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7jxNhhjb5ZOt",
        "colab_type": "text"
      },
      "source": [
        "And this is our final look of the data."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "k5HOIIzniFzz",
        "colab_type": "code",
        "outputId": "5776a696-82b3-4ab0-b445-b8c07cfdc20b",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 185
        }
      },
      "source": [
        "processed_review.select('final').limit(5).show(truncate=90)"
      ],
      "execution_count": 39,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+------------------------------------------------------------------------------------------+\n",
            "|                                                                                     final|\n",
            "+------------------------------------------------------------------------------------------+\n",
            "|                                                                                        []|\n",
            "|[exactly, quite, even, expectedas, add, one, carry, small, hint, grape, buy, put, next,...|\n",
            "|[job, device, would, otherwise, pop, allow, reduction, high, frequency, cloth, block, l...|\n",
            "|[nice, windscreen, protect, mic, prevent, thing, gooseneck, able, hold, require, carefu...|\n",
            "|[filter, great, look, perform, studio, youre, eliminate, pop, record, sing, pop_filter,...|\n",
            "+------------------------------------------------------------------------------------------+\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "YkTzRKEzaYLM",
        "colab_type": "text"
      },
      "source": [
        "There are additional ways how we could provide cleaner data with Spark NLP for topic modelling analysis. For example: \n",
        "\n",
        "* You might want to incorporate [**SentenceDetector**](https://nlp.johnsnowlabs.com/docs/en/annotators#sentencedetector) in order to ignore n-grams on the borders of sentences since tokenization in Spark NLP does not account for sentence borders. \n",
        "\n",
        "* [**DependencyParser**](https://nlp.johnsnowlabs.com/docs/en/annotators#dependency-parsers) also could be used to provide more meaningful n-grams, namely the ones with dependency relation.\n",
        "\n",
        "* Spell checker also could be incorporated at the early steps of the NLP pipeline for less noisy results. There are various options in Spark NLP such as [**NorvigSpellChecker**](https://nlp.johnsnowlabs.com/docs/en/annotators#norvig-spellchecker) and [**SymmetricSpellChecker**](https://nlp.johnsnowlabs.com/docs/en/annotators#symmetric-spellchecker).\n",
        "\n",
        "However, in this tutorial we will omit these options since they probably will not bring significant improvements fot topic modelling."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-OTK4DVHxcR0",
        "colab_type": "text"
      },
      "source": [
        "## Vectorization "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "OP4HlvFR5jXp",
        "colab_type": "text"
      },
      "source": [
        "Now we are set to vectorization of our data. First, we will proceed with **TF** (*term frequency*) vectorization with **CountVectorizer** in PySpark. We fit tf dictionary and then transform the data to vectors of counts."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "tqGT4ss7r_w2",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.feature import CountVectorizer\n",
        "\n",
        "tfizer = CountVectorizer(inputCol='final', outputCol='tf_features')\n",
        "tf_model = tfizer.fit(processed_review)\n",
        "tf_result = tf_model.transform(processed_review)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "217ZJrT76B2h",
        "colab_type": "text"
      },
      "source": [
        "After we get TF results, we can account for words that are frequent for all the documents. We can use **IDF** (*inverse document frequency*) to lower score of such words."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ox3dEUoyx2Ss",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.feature import IDF\n",
        "\n",
        "idfizer = IDF(inputCol='tf_features', outputCol='tf_idf_features')\n",
        "idf_model = idfizer.fit(tf_result)\n",
        "tfidf_result = idf_model.transform(tf_result)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ezlCICGwzvQu",
        "colab_type": "text"
      },
      "source": [
        "## LDA"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kelkPspY6okR",
        "colab_type": "text"
      },
      "source": [
        "Finally, we are ready to model topics in our data with [**LDA**](https://spark.apache.org/docs/2.2.0/ml-clustering.html#latent-dirichlet-allocation-lda) (*Latent Dirichlet Allocation*). To use the algorithm, we have to provide the number of topics we presume our data contains and the number of iterations for the LDA algorithm. Then, we initialize the model and train it."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ID3lf4GjzxJq",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.clustering import LDA\n",
        "\n",
        "num_topics = 6\n",
        "max_iter = 10\n",
        "\n",
        "lda = LDA(k=num_topics, maxIter=max_iter, featuresCol='tf_idf_features')\n",
        "lda_model = lda.fit(tfidf_result)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "DslMpvbT7GXQ",
        "colab_type": "text"
      },
      "source": [
        "To be able to see words that characterize the defined topics, we need to convert word ids into actual words with the custom function. This function will again be converted to PySpark UDF to be used on our topic dataframe."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "OxbhwXnQ0dEa",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "vocab = tf_model.vocabulary\n",
        "\n",
        "def get_words(token_list):\n",
        "     return [vocab[token_id] for token_id in token_list]\n",
        "       \n",
        "udf_to_words = F.udf(get_words, T.ArrayType(T.StringType()))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "GllCV-hw7fgL",
        "colab_type": "text"
      },
      "source": [
        "Let's define the number of top words per topic we would like to see and extract the words with our function."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Fhsk4J5p0lOm",
        "colab_type": "code",
        "outputId": "e8e5edc3-8b63-47e1-dfcb-9aba290208e0",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 202
        }
      },
      "source": [
        "num_top_words = 10\n",
        "\n",
        "topics = lda_model.describeTopics(num_top_words).withColumn('topicWords', udf_to_words(F.col('termIndices')))\n",
        "topics.select('topic', 'topicWords').show(truncate=90)"
      ],
      "execution_count": 44,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+-----+------------------------------------------------------------------------------------+\n",
            "|topic|                                                                          topicWords|\n",
            "+-----+------------------------------------------------------------------------------------+\n",
            "|    0|             [pick, bow, shockmount, store, pm, use, mic, tortex_pick, nice, guitar]|\n",
            "|    1|                  [pedal, amp, great, sound, tuner, good, cable, use, effect, price]|\n",
            "|    2|                     [string, use, guitar, sound, one, strap, get, well, work, good]|\n",
            "|    3|       [mic, use, guitar, good, word, great, quality, word_word, work, good_quality]|\n",
            "|    4|                    [use, string, pick, pedal, one, good, buy, guitar, sound, great]|\n",
            "|    5|[guitar, harmonica, yamaha, use, volume_pedal, string, record, tuner, battery, work]|\n",
            "+-----+------------------------------------------------------------------------------------+\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "CcwVxupn7wKe",
        "colab_type": "text"
      },
      "source": [
        "And that's it! We are done with topic modelling pipeline on review data. I hope this tutorial was somehow helpful for you. \n",
        "\n",
        "Good luck with your own experiments!\n",
        "\n"
      ]
    }
  ]
}