1_3_NLP_on_Apache_Spark.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "1.3_NLP_on_Apache_Spark.ipynb",
      "provenance": [],
      "toc_visible": true,
      "authorship_tag": "ABX9TyPyFaFiU0BTzUWEmuo57HPB",
      "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/alexander-n-thomas/NLU_tutorial/blob/master/1_3_NLP_on_Apache_Spark.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "d4zpDO5R1ox4",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 390
        },
        "outputId": "2c926d51-80ba-48e9-e8e1-22de507159e3"
      },
      "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.4\n",
        "\n",
        "# Install Spark NLP\n",
        "! pip install --ignore-installed spark-nlp==2.5.1"
      ],
      "execution_count": 2,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "openjdk version \"1.8.0_252\"\n",
            "OpenJDK Runtime Environment (build 1.8.0_252-8u252-b09-1~18.04-b09)\n",
            "OpenJDK 64-Bit Server VM (build 25.252-b09, mixed mode)\n",
            "Collecting pyspark==2.4.4\n",
            "\u001b[?25l  Downloading https://files.pythonhosted.org/packages/87/21/f05c186f4ddb01d15d0ddc36ef4b7e3cedbeb6412274a41f26b55a650ee5/pyspark-2.4.4.tar.gz (215.7MB)\n",
            "\u001b[K     |████████████████████████████████| 215.7MB 54kB/s \n",
            "\u001b[?25hCollecting py4j==0.10.7\n",
            "\u001b[?25l  Downloading https://files.pythonhosted.org/packages/e3/53/c737818eb9a7dc32a7cd4f1396e787bd94200c3997c72c1dbe028587bd76/py4j-0.10.7-py2.py3-none-any.whl (197kB)\n",
            "\u001b[K     |████████████████████████████████| 204kB 38.9MB/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-2.4.4-py2.py3-none-any.whl size=216130388 sha256=56fcd880c22a56a2a8b688b104499ba6c1617378899742fb5b0874ece67ddb96\n",
            "  Stored in directory: /root/.cache/pip/wheels/ab/09/4d/0d184230058e654eb1b04467dbc1292f00eaa186544604b471\n",
            "Successfully built pyspark\n",
            "Installing collected packages: py4j, pyspark\n",
            "Successfully installed py4j-0.10.7 pyspark-2.4.4\n",
            "Collecting spark-nlp==2.5.1\n",
            "\u001b[?25l  Downloading https://files.pythonhosted.org/packages/df/b4/db653f8080a446de8ce981b262d85c85c61de7e920930726da0d1c6b4c65/spark_nlp-2.5.1-py2.py3-none-any.whl (121kB)\n",
            "\u001b[K     |████████████████████████████████| 122kB 5.2MB/s \n",
            "\u001b[?25hInstalling collected packages: spark-nlp\n",
            "Successfully installed spark-nlp-2.5.1\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "HzgTx-7OuEnm",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "! mkdir -p data"
      ],
      "execution_count": 4,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "dgBCSO9Fr-M5",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 212
        },
        "outputId": "6ecefdc7-4cb1-4233-e898-0f397ff35813"
      },
      "source": [
        "! wget https://archive.ics.uci.edu/ml/machine-learning-databases/20newsgroups-mld/mini_newsgroups.tar.gz"
      ],
      "execution_count": 5,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "--2020-07-23 04:05:28--  https://archive.ics.uci.edu/ml/machine-learning-databases/20newsgroups-mld/mini_newsgroups.tar.gz\n",
            "Resolving archive.ics.uci.edu (archive.ics.uci.edu)... 128.195.10.252\n",
            "Connecting to archive.ics.uci.edu (archive.ics.uci.edu)|128.195.10.252|:443... connected.\n",
            "HTTP request sent, awaiting response... 200 OK\n",
            "Length: 1860687 (1.8M) [application/x-httpd-php]\n",
            "Saving to: ‘mini_newsgroups.tar.gz’\n",
            "\n",
            "mini_newsgroups.tar 100%[===================>]   1.77M  8.87MB/s    in 0.2s    \n",
            "\n",
            "2020-07-23 04:05:29 (8.87 MB/s) - ‘mini_newsgroups.tar.gz’ saved [1860687/1860687]\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "B7DAiInZuIsG",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "! tar xzf mini_newsgroups.tar.gz -C ./data/"
      ],
      "execution_count": 6,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "nAMhZwEW4M3A",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 212
        },
        "outputId": "1cd10ec4-ec0b-49df-cb8e-a278428f3fb8"
      },
      "source": [
        "! wget https://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data"
      ],
      "execution_count": 23,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "--2020-07-23 04:15:00--  https://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data\n",
            "Resolving archive.ics.uci.edu (archive.ics.uci.edu)... 128.195.10.252\n",
            "Connecting to archive.ics.uci.edu (archive.ics.uci.edu)|128.195.10.252|:443... connected.\n",
            "HTTP request sent, awaiting response... 200 OK\n",
            "Length: 4551 (4.4K) [application/x-httpd-php]\n",
            "Saving to: ‘iris.data’\n",
            "\n",
            "\riris.data             0%[                    ]       0  --.-KB/s               \riris.data           100%[===================>]   4.44K  --.-KB/s    in 0s      \n",
            "\n",
            "2020-07-23 04:15:00 (73.1 MB/s) - ‘iris.data’ saved [4551/4551]\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "-8N8Uqqj4MnM",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "! mv iris.data ./data"
      ],
      "execution_count": 24,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "vTLcAwbL-q_3",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 123
        },
        "outputId": "0b24a671-5d10-4ca6-829e-1898cc235ec6"
      },
      "source": [
        "! wget https://github.com/alexander-n-thomas/spark-nlp-book-prod/raw/master/en_lemmas.txt"
      ],
      "execution_count": 74,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "--2020-07-23 04:43:12--  https://github.com/alexander-n-thomas/spark-nlp-book-prod/raw/master/en_lemmas.txt\n",
            "Resolving github.com (github.com)... 140.82.113.3\n",
            "Connecting to github.com (github.com)|140.82.113.3|:443... connected.\n",
            "HTTP request sent, awaiting response... 404 Not Found\n",
            "2020-07-23 04:43:13 ERROR 404: Not Found.\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "YofQ9lTttiGS",
        "colab_type": "text"
      },
      "source": [
        "## NLP on Apache Spark\n",
        "It's no longer news that there is a data deluge. Every day, people and devices are creating huge amounts of data. Text data is definitely one of the main kinds of data that humans produce. People write millions of comments, product reviews, Reddit messages, and tweets per day. This data is incredibly valuable—for both research and commerce. Because of the scale at which this data is created, our approach to working with it has changed.\n",
        "\n",
        "Most of the original research in NLP was done on small data sets with hundreds or thousands of documents. You may think that it would be easier to build NLP applications now that we have so much more text data with which to build better models. However, these pieces of text have different pragmatics and are of different varieties, so leveraging them is more complicated from a data-science perspective. From the software engineering perspective, big data introduces many challenges. Structured data has predictable size and organization, which makes it easier to store and distribute efficiently. Text data is much less consistent. This makes parallelizing and distributing work more important and potentially more complex. Distributed computing frameworks like Spark help us manage these challenges and complexities.\n",
        "\n",
        "In this chapter, we will discuss the Apache Spark and Spark NLP. First, we will cover some basic concepts that will help us understand distributed computing. Then, we will talk briefly about the history of distributed computing. We will talk about some important modules in Spark—Spark SQL and MLlib. This will give us the background and context needed to talk about Spark NLP in technical detail.\n",
        "\n",
        "Now, we'll cover some technical concepts that will be helpful in understanding how Spark works. The explanations will be high level. If you are interested in maximizing performance, I suggest looking more into these topics. For the general audience, I hope this material will give you the intuition necessary to help make decisions when designing and building Spark-based applications."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "VYAKO09Ztj1N",
        "colab_type": "text"
      },
      "source": [
        "## Parallelism, Concurrency, Distributing Computation\n",
        "Let's start by defining some terms.  A process can be thought of as a running program.  A process executes its code using an allotted portion of memory also known as a memory space. A thread is a sequence of execution steps within a process that the operating system can schedule. Sharing data between processes generally requires copying data between the different memory spaces.  When a Java or Scala program is run, the Java Virtual Machine (JVM) is the process. The threads of a process share access to the same memory space, which they access concurrently.\n",
        "\n",
        " Concurrent access of data can be tricky. For example, let's say we want to generate word counts. If two threads are working on this process, it's possible for us to get the wrong count. Consider the following program (written in pseudo-Python). In this program, we will use a thread pool.  A thread pool is a way to separate partitioning work from scheduling. We allocate a certain number of threads, and then we go through our data asking for threads for the pool. The operating system can then schedule the work.\n",
        "\n",
        "```\n",
        "0:  def word_count(tokenized_documents): # list of lists of tokens\n",
        "1:      word_counts = {}\n",
        "2:      thread_pool = ThreadPool()\n",
        "3:      i = 0\n",
        "4:      for thread in thread_pool\n",
        "5:          run thread:\n",
        "6:              while i < len(tokenized_documents):\n",
        "7:                  doc = tokenized_documents[i]\n",
        "8:                  i += 1\n",
        "9:                  for token in doc:\n",
        "10:                     old_count = word_counts.get(token, 0)\n",
        "11:                     word_counts[token] = old_count + 1\n",
        "12:     return word_counts\n",
        "```"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ox34D_1Sto1f",
        "colab_type": "text"
      },
      "source": [
        "This looks reasonable, but we see that the code under `run thread` references data in the shared memory space like `i` and `word_counts`. tableone shows the execution of this program with two `thread`s in the `ThreadPool` starting at line 6.\n",
        "\n",
        "_Two `threads` in the `ThreadPool`_\n",
        "\n",
        "time|thread1|thread2|i|valid_state\n",
        "----|-------|-------|-|-----------\n",
        "0|while i < len(tokenized_documents)||0|yes\n",
        "1||while i < len(tokenized_documents)|0|yes\n",
        "2|doc = tokenized_documents[i]||0|yes\n",
        "3||doc = tokenized_documents[i]|0|NO\n",
        "\n",
        "At time 3, `thread2` will be retrieving `tokenized_documents[0]`, while `thread1` is already set to work on the first document.  This program has a race condition, in which we can get incorrect results depending on the sequence of operations done in the different threads. Avoiding these problems generally involves writing code that is safe for concurrent access. For example, we can pause `thread2` until `thread1` is finished updating by  locking on `tokenized_documents`. If you look at the code, there is another race condition, on `i`. If thread1 takes the last document, `tokenized_documents[N-1]`, `thread2` starts its while-loop check, `thread1` updates `i`, then `thread2` uses `i`. We will be accessing `tokenized_documents[N]`, which doesn't exist. So let's lock on `i`."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "sVdEHp-ltpCk",
        "colab_type": "text"
      },
      "source": [
        "```\n",
        "0:  def word_count(tokenized_documents): # list of lists of tokens\n",
        "1:      word_counts = {}\n",
        "2:      thread_pool = ThreadPool()\n",
        "3:      i = 0\n",
        "4:      for thread in thread_pool\n",
        "5:          run thread:\n",
        "6:              while True:\n",
        "7:                  lock i:\n",
        "8:                      if i < len(tokenized_documents)\n",
        "9:                          doc = tokenized_documents[i]\n",
        "10:                         i += 1\n",
        "11:                     else:\n",
        "12:                         break\n",
        "13:                 for token in doc:\n",
        "14:                     lock word_counts:\n",
        "15:                         old_count = word_counts.get(token, 0)\n",
        "16:                         word_counts[token] = old_count + 1\n",
        "17:     return word_counts\n",
        "```\n",
        "\n",
        "Now, we are locking on `i` and checking `i` in the loop. We also lock on `word_counts` so that if two threads want to update the counts of the same word, they won't accidentally pull a stale value for `old_count`.\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "QJU_0w8atp3R",
        "colab_type": "text"
      },
      "source": [
        "time|thread1|thread2|i|valid_state\n",
        "----|-------|-------|-|-----------\n",
        "0|lock i||0|yes\n",
        "1|if i < len(tokenized_documents)|blocked|0|yes\n",
        "2|doc = tokenized_documents[i]|blocked|0|yes\n",
        "3|i += 1|blocked|0|yes\n",
        "4||lock i|1|yes\n",
        "5|lock word_counts||1|yes\n",
        "6||if i < len(tokenized_documents)|1|yes\n",
        "7|old_count = word_counts.get(token, 0)||1|yes\n",
        "8||doc = tokenized_documents[i]|1|yes\n",
        "9|word_counts[token] = old_count + 1||1|yes\n",
        "10||i += 1|1|yes\n",
        "11|lock word_counts||2|yes\n",
        "12|old_count = word_counts.get(token, 0)|blocked|2|yes\n",
        "13|word_counts[token] = old_count + 1|blocked|2|yes\n",
        "14||lock word_counts|2|yes\n",
        "15|blocked\t|old_count = word_counts.get(token, 0)|2|yes\n",
        "16|blocked\t|word_counts[token] = old_count + 1|2|yes\n",
        "\n",
        "We fixed the problem, but at the cost of frequently blocking one of the threads. This means that we are getting less advantage of the parallelism. It would be better to design our algorithm so that the threads don't share state. We will see an example of this when we talk about MapReduce. \n",
        "\n",
        "Sometimes, parallelizing on one machine is not sufficient, so we distribute the work across many machines grouped together in a cluster. When all the work is done on a machine, we are bringing the data (in memory or on disk) to the code, but when distributing work we are bringing the code to the data. Distributing the work of a program across a cluster means we have new concerns. We don't have access to a shared memory space, so we need to be more thoughtful in how we design our algorithms. Although processes on different machines don't share a common memory space, we still need to consider concurrency because the threads of the process on a given machine of the cluster still share common (local) memory space. Fortunately, modern frameworks like Spark mostly take care of these concerns, but it's still good to keep this in mind when designing your programs.\n",
        "\n",
        "Programs that work on text data often find some form of parallelization helpful because processing the text into structured data is often the most time-consuming stage of a program. Most NLP pipelines ultimately output structured numeric data, which means that the data that's loaded, the text, can often be much larger than the data that is output. Unfortunately, because of the complexity of NLP algorithms, text processing in distributed frameworks is generally limited to basic techniques. Fortunately, we have Spark NLP, which we will discuss shortly.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "4zracfS7tqtI",
        "colab_type": "text"
      },
      "source": [
        "### Parallelization Before Apache Hadoop\n",
        "HTCondor is a framework developed at the University of Wisconsin–Madison starting in 1988. It boasts an impressive catalog of uses. It was used by NASA, the Human Genome Project, and the Large Hadron Collider. Technically, it's not just a framework for distributing computation—it also can manage resources. In fact, it can be used with other frameworks for distributing computation. It was built with the idea that machines in the cluster may be owned by different users, so work can be scheduled based on available resources. This is from a time when clusters of computers were not as available.\n",
        "\n",
        "GNU parallel and pexec are UNIX tools that can be used to parallelize work on a single machine, as well as across machines. This requires that the distributable portions of work be run from the command line. These tools allow us to utilize resources across machines, but it doesn't help with parallelizing our algorithms.\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "9G1uNCa_tq6p",
        "colab_type": "text"
      },
      "source": [
        "### MapReduce and Apache Hadoop\n",
        "We can represent distributed computation with two operations: map and reduce. The map operation can be used to transform, filter, or sort data. The reduce operation can be used to group or summarize data. Let's return to our word count example to see how we can use these two operations for a basic NLP task.\n",
        "\n",
        "```\n",
        "def map(docs):\n",
        "    for doc in docs:\n",
        "        for token in doc:\n",
        "            yield (token, 1)\n",
        "\n",
        "def reduce(records):\n",
        "    word_counts = {}\n",
        "    for token, count in records:\n",
        "        word_counts[token] = word_counts.get(token, 0) + count\n",
        "    for word, count in word_counts.items():\n",
        "        yield (word, count)\n",
        "```\n",
        "\n",
        "The data is loaded in partitions, with some documents going to each `mapper` process on the cluster. There can be multiple `mapper`s per machine. Each `mapper` runs the map function on their documents and saves the results to the disk. After all of the `mapper`s have completed, the data from the `mapper` stage is shuffled so that, all the records with the same key (`word` in this case), are in the same partition. This data is now sorted so that within a partition, all the records are ordered by key. Finally, the sorted data is loaded and the `reduce` step is called for each partition `reducer` process combining all of the counts. In between stages, the data is saved to the disk.\n",
        "\n",
        "MapReduce can express most distributed algorithms, but some are difficult or downright awkward in this framework. This is why abstractions to MapReduce were developed rather quickly.\n",
        "\n",
        "Apache Hadoop is the popular open source implementation of MapReduce along with a distributed file system,  Hadoop Distributed File System (HDFS). To write a Hadoop program, you need to select or define an input format, mapper, reducer, and output format. There have been many libraries and frameworks to allow higher levels of implementing a program.\n",
        "\n",
        " Apache Pig is a framework for expressing MapReduce programs in procedural code. Its procedural nature makes implementing extract, transform, load (ETL) programs very convenient and straightforward. However, other types of programs, model training programs for example, are much more difficult.  The language that Apache Pig uses is called Pig Latin. There is some overlap with SQL, so if someone knows SQL well, learning Pig Latin is easy.\n",
        "\n",
        " Apache Hive is a data warehousing framework originally built on Hadoop. Hive allows users to write SQL that is executed with MapReduce. Now, Hive can run using other distributed frameworks in addition to Hadoop, including Spark.\n",
        "\n",
        "### Apache Spark\n",
        "Spark is a project started by Matei Zaharia. Spark is a distributed computing framework. There are important differences in how Spark and Hadoop process data. Spark allows users to write arbitrary code against distributed data. Currently, there are official APIs for Spark in Scala, Java, Python, and R. Spark does not save intermediate data to disk. Usually, Spark-based programs will keep data in memory, though this can be changed by configuration to also utilize the disk. This allows for quicker processing but can require more scale out (more machines), or more scale up (machines with more memory).\n",
        "\n",
        "Let's look at `word_count` in Spark."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "s2VtYbneyysy",
        "colab_type": "text"
      },
      "source": [
        "## Architecture of Apache Spark\n",
        "Spark is organized around a driver that is running the program, a master that manages resources and distributes work, and workers that execute computation. There are a number of possible masters. Spark ships with its own master, which is what is used in standalone and local modes. You can also use Apache YARN or Apache Mesos. In my experience, Apache YARN is the most common choice for enterprise systems.\n",
        "\n",
        "Let's take a more detailed look at Spark's architecture.\n",
        "\n",
        "### Physical Architecture\n",
        "We start our program on the submitting machine which submits an application. This driver runs the application on the client machine and sends jobs to the  spark master to be distributed to the workers. The spark master may not be a completely separate machine. That machine may also be doing work on the cluster and so would be a worker as well. Also, you may be running your program on the spark master, and so it could also be the client machine.\n",
        "\n",
        "There are two modes that you can use to start a Spark application: cluster mode and client mode. If the machine submitting the application is the same machine that runs the application, you are in client mode, because you are submitting from the client machine. Otherwise, you are in cluster mode. Generally, you use client mode if your machine is inside the cluster and cluster mode if it is not (see Figures #client and #cluster).\n",
        "\n",
        "![Physical architecture (client mode)](https://i.imgur.com/ZiuOmLo.png)  \n",
        "_Physical architecture (client mode)_\n",
        "\n",
        "![Physical architecture (cluster mode)](https://i.imgur.com/lMltdz3.png)  \n",
        "_Physical architecture (cluster mode)_\n",
        "\n",
        "You can also run Spark in local mode in which, as the name implies, client machine, spark master, and worker are all the same machine. This is very useful for developing and testing Spark applications. It is also useful if you want to parallelize work on one machine.\n",
        "\n",
        "Now that we have an idea of the physical architecture, let's look at the logical architecture."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "BG0sLPj3yzI_",
        "colab_type": "text"
      },
      "source": [
        "### Logical Architecture\n",
        "\n",
        "In looking at the logical architecture of Spark, we will treat the client machine, spark master, and worker as if they are different (see architecture). The driver is a JVM process that will submit work to the spark master. If the program is a Java or Scala program, then it is also the process running the program. If the program is in Python or R, then the driver process is a separate process from that running the program.\n",
        "\n",
        "![Logical architecture](https://i.imgur.com/WaM6zvN.png)  \n",
        "_Logical architecture_\n",
        "\n",
        "The JVM processes on the workers are called executors. The work to be done is defined on the driver and submitted to the spark master, which orchestrates the executors to do the work. The next step in understanding Spark is understanding how the data is distributed.\n",
        "\n",
        "#### RDDs\n",
        "\n",
        "Spark distributes data in resilient distributed datasets (RDDs). RDDs allow users to work with distributed data almost as if it were a collection located in the driver."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "4sjUgZQXyzNn",
        "colab_type": "text"
      },
      "source": [
        "#### Partitioning\n",
        "\n",
        "In Spark, data is partitioned across the cluster. There are usually more partitions than executors. This allows for each thread on each executor to be utilized. Spark will distribute the data in the RDD evenly across the cluster into the default number of partitions. We can specify the number of partitions, and we can specify a field to partition by. This can be very useful when your algorithm requires some degree of locality—for example, having all the tweets from a single user.\n",
        "\n",
        "#### Serialization\n",
        "Any code that is shipped to the data should refer only to serializable objects. `NotSerializableException` errors are common and can be nearly inscrutable to those new to Spark.  When we map over an `RDD`, we are creating a `Function` and sending it to the machines with the data. A function is the code that defines it, and the data needed in the definition. This second part is called the closure of the function. Identifying what objects are needed for a function is a complicated task, and sometimes extraneous objects can be captured. If you are having problems with serializability, there are a couple of possible solutions. The following are questions that can help you find the right solution:\n",
        "\n",
        "* Are you using your own custom classes? Make sure that they are serializable.\n",
        "* Are you loading a resource? Perhaps your distributed code should load it lazily so that it is loaded on each executor, instead of being loaded on the driver and shipped to the executors.\n",
        "* Are Spark objects (`SparkSession`, `RDD`s) being captured in a closure? This can happen when you are defining your function anonymously. If your function is defined anonymously, perhaps you can define it elsewhere.\n",
        "\n",
        "These tips can help find common errors, but the solution to this problem is something that can be determined only on a case-by-case basis."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "L32XtqkiyzSS",
        "colab_type": "text"
      },
      "source": [
        "#### Ordering\n",
        "\n",
        "When working with distributed data, there is not necessarily a guaranteed order to the items in your data. When writing your code, keep in mind that the data exists in partitions across the cluster.\n",
        "\n",
        "This is not to say that we cannot define an order. We can define an order on the partitions by an index. In this situation, the \"first\" element of the `RDD` will be the first element of the first partition. Furthermore, let's say we want to order an `RDD[Int]` ascending by value. Using our ordering on partitions, we can shuffle the data such that all elements in partition `i` are less than all elements in partition `i+1`. From here, we can sort each partition. Now we have a sorted `RDD`. This is an expensive operation, however.\n",
        "\n",
        "#### Output and logging\n",
        "\n",
        "When writing functions that are used to transform data, it is often useful to print statements or, preferably, to log statements to look at the state of variables in the function. In a distributed context, this is more complicated because the function is not running on the same machine as the program. Accessing the logs and the stdout generally depends on the configuration of the cluster and which master you are using. In some situations, it may suffice to run your program on a small set of data in local mode.\n",
        "\n",
        "#### Spark jobs\n",
        "\n",
        "A Spark-based program will have a `SparkSession`, which is how the driver talks to the master.  Before Spark version 2.x, the `SparkContext` was used for this. There is still a `SparkContext`, but it is part of the `SparkSession` now. This `SparkSession` represents the `App`. The `App` submits `jobs` to the master. The `jobs` are broken into `stages`, which are logical groupings of work in the job. The `stages` are broken into `tasks`, which represent the work to be done on each partition.\n",
        "\n",
        "![Spark jobs](https://i.imgur.com/Lmfc5Nc.png)  \n",
        "_Spark jobs_\n",
        "\n",
        "Not every operation on data will start a job. Spark is lazy—in a good way. Execution is done only when a result is needed. This allows the work done to be organized into an execution plan to improve efficiency. There are certain operations, sometimes referred to as actions, that will cause execution immediately. These are operations that return specific values, e.g. `aggregate`. There are also operations where further execution planning becomes impossible until computed, e.g. `zipWithIndex`.\n",
        "\n",
        "  This execution plan is often referred to as the Directed Acyclic Graph (DAG). Data in Spark is defined by its DAG, which includes its sources and whatever operations were run to generate it. This allows Spark to remove data from memory as necessary without losing reference to the data. If data that was generated and removed is referred to later, it will be regenerated. This can be time-consuming. Fortunately, we can instruct Spark to keep data if we need to."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kVDozVzNyzWE",
        "colab_type": "text"
      },
      "source": [
        "#### Persisting\n",
        "\n",
        "The basic way of persisting data in Spark is with the persist method. This creates a checkpoint. You can also use the cache method and provide options for configuring how the data will be persisted. The data will still be generated lazily, but once it is generated, it will be kept.\n",
        "\n",
        "Let's look at an example:"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "YT9sA32I2uxJ",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from operator import concat, itemgetter, methodcaller\n",
        "import os\n",
        "from time import sleep\n",
        "\n",
        "import pyspark\n",
        "from pyspark import SparkConf\n",
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql import functions as fun\n",
        "from pyspark.sql.types import *"
      ],
      "execution_count": 9,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "9hk7cOg82wxl",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "packages = ','.join([\n",
        "    \"com.johnsnowlabs.nlp:spark-nlp_2.11:2.4.5\",\n",
        "])\n",
        "\n",
        "spark = SparkSession.builder \\\n",
        "    .master(\"local[*]\") \\\n",
        "    .appName(\"spark-nlp-book-p1c3\") \\\n",
        "    .config(\"spark.jars.packages\", packages)\\\n",
        "    .getOrCreate()"
      ],
      "execution_count": 10,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "qTnaiFGd2zr2",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "def has_moon(text):\n",
        "    if 'moon' in text:\n",
        "        sleep(1)\n",
        "        return True\n",
        "    else:\n",
        "        return False"
      ],
      "execution_count": 11,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "15Rps9or1UgZ",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 123
        },
        "outputId": "7697f362-c7e7-4b78-a5aa-86416bf4847d"
      },
      "source": [
        "# RDD containing filepath-text pairs\n",
        "path = os.path.join('data', 'mini_newsgroups', 'sci.space')\n",
        "text_pairs = spark.sparkContext\\\n",
        "    .wholeTextFiles(path) \n",
        "texts = text_pairs.map(itemgetter(1))\n",
        "lower_cased = texts.map(methodcaller('lower'))\n",
        "moon_texts = texts.filter(has_moon).persist()\n",
        "print('This appears quickly because the previous operations are '\n",
        "      'all lazy')\n",
        "print(moon_texts.count())\n",
        "print('This appears slowly since the count method will call '\n",
        "      'has_moon which sleeps')\n",
        "print(moon_texts.reduce(concat)[:100])\n",
        "print('This appears quickly because has_moon will not be '\n",
        "      'called due to the data being persisted')"
      ],
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "This appears quickly because the previous operations are all lazy\n",
            "11\n",
            "This appears slowly since the count method will call has_moon which sleeps\n",
            "Xref: cantaloupe.srv.cs.cmu.edu sci.space:60944 sci.astro:34770 alt.sci.planetary:1193\n",
            "Path: cantalo\n",
            "This appears quickly because has_moon will not be called due to the data being persisted\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "rItFYkup28xs",
        "colab_type": "text"
      },
      "source": [
        "Now that we have an idea of how Spark works, let's go back to our word-count problem."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "aeQtbQ2_1Wll",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from collections import Counter\n",
        "from operator import add\n",
        "\n",
        "from nltk.tokenize import RegexpTokenizer"
      ],
      "execution_count": 13,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "kvb4Iavz2_1z",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 107
        },
        "outputId": "c32035d9-024a-4ad3-f43c-eee6fd469990"
      },
      "source": [
        "# RDD containing filepath-text pairs\n",
        "texts = spark.sparkContext.wholeTextFiles(path) \n",
        "print('\\n\\nfilepath-text pair of first document')\n",
        "print(texts.first())"
      ],
      "execution_count": 14,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "\n",
            "\n",
            "filepath-text pair of first document\n",
            "('file:/content/data/mini_newsgroups/sci.space/61534', \"Path: cantaloupe.srv.cs.cmu.edu!das-news.harvard.edu!noc.near.net!uunet!digex.com!digex.net!not-for-mail\\nFrom: prb@access.digex.net (Pat)\\nNewsgroups: sci.space\\nSubject: Re: Space Manuevering Tug (was HST servicing mission_)\\nDate: 1 May 1993 15:25:38 -0400\\nOrganization: Express Access Online Communications USA\\nLines: 83\\nMessage-ID: <1ruiri$dig@access.digex.net>\\nReferences: <1rnaih$jvj@access.digex.net> <C6BBow.IH9@news.cso.uiuc.edu>\\nNNTP-Posting-Host: access.digex.net\\n\\nIn article <C6BBow.IH9@news.cso.uiuc.edu> jbh55289@uxa.cso.uiuc.edu (Josh Hopkins) writes:\\n>prb@access.digex.net (Pat) writes:\\n|\\n|>Given that what i described for the HST  seemed to be the SMT,   and given\\n|>the mass amrgins on the discovery mission  is tight enough that  spacewalking\\n|>has to be carefully constrained.....  No EDO pallets,  no spare Suits,\\n|>no extra MMU's.   \\n|\\n|Has someone actually verified that mass is the predominant constraint on this\\n|mission?  You seem to be assuming it without giving supporting evidence.  \\n|\\n\\nSomeone from NASA posted that there were very significant mass margins\\non the HST re-boost mission.  A while back i had asked why not carry\\nthe EDO pallet up,  and the answer was the mass margins were tight enough, they weren't even carrying extra suits.\\n\\n|>WHy not do this?\\n|\\n|>\\tQuick Test  Goldins philosophjy  of faster cheaper, better.\\n|\\n|>Build a real fast Space TUg,  to handle the re-boost  of the HST  using\\n|>clean Cryo fuels,  and get it ready before the  HST mission.\\n|\\n|Pat, this would be slower, more expensive and worse.  \\n|\\n|Slower:  The shuttle mission is scheduled to go up in December.  That's less\\n|than eight months away.  There is no way you could build new hardware, retrain\\n|and reschedule the EVA's in that time.\\n|\\n\\nWhere's wingo when you need him:-)\\n\\nCOme on.   Knock that S**T off.\\n\\nYOu forget,  that during skylab,  they did  overnight mission planning\\nfor the repair EVA's.   Also during the��   \\nIntelsat Mission,   they did overnight  WETF simulations.\\nI somehow think they could train up a new  EVA in  8 months.\\n\\nAnd as for building hardware,  anything can  be built if you want it\\nbad enough.\\n\\nYOu forget,  the  BUS 1  is already built.  all they'd ahve to do\\nis soup it up, even test it  on a delta mission.\\n\\nDon't get into this mode of  negativism.   besides,  at the rate\\nmissions slip,   the Discovery won't launch on this mission until\\nMarch.  that's almost a year.\\n\\n|More Expensive:  Your proposal still requires the shuttle to do everything it\\n|was going to do execpt fire the OMS.  In addition, you've added significant\\n|extra cost for a new piece of complex hardware.\\n|\\n\\nAh,  but how much more expensive is the Second HST servicing mission.\\n\\nYOu forget,  there is a bum FGS,  the Solar array electronics, are\\ngetting hinky  and there is still 8 months until the servicing mission.\\n\\nThe time for the space walks are growing rapidly.  THis was orignally\\nplanned out as 3 spacewalks,  now they are at 5 EVA's  with 3 reserve\\nwalks.\\n\\nIf the SMT can avoid a second servicing mission that's $500 million\\nsaved.  If the Weight savings,  means they  can sit on orbit  for 30 Days.\\nand  handle any contingency  problems,  that's quite a savings.\\n\\n|\\n|According to a GAO report on the OMV I have before me, there are\\n|only two currently planned missions that could use such a vehicle -- HST and\\n|AXAF.  Since AXAF has since been scaled back and HST can rely on the shuttle,\\n|there doesn't seem to be any need for your vehicle.\\n\\n\\nOf course,  there wasn't any need for the Saturn  V  after apollo too.\\n\\nas for the problems with the aperture door,  I am sure they can\\nwork out some way to handle  that.  Maybe a Plug  made from\\nFrozen ice.?   it'll keep out any contamination,\\nyet sublime away  after teh boost.\\n\\npat\\n\\n\")\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "99l4bgDe3BV5",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 107
        },
        "outputId": "ac24662f-5601-4b02-f509-14a47684f2cf"
      },
      "source": [
        "tokenizer = RegexpTokenizer(r'\\w+', gaps=False)\n",
        "tokenized_texts = texts.map(\n",
        "    lambda path_text: tokenizer.tokenize(path_text[1]))\n",
        "print('\\n\\ntokenized text of first document')\n",
        "print(tokenized_texts.first())"
      ],
      "execution_count": 15,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "\n",
            "\n",
            "tokenized text of first document\n",
            "['Path', 'cantaloupe', 'srv', 'cs', 'cmu', 'edu', 'das', 'news', 'harvard', 'edu', 'noc', 'near', 'net', 'uunet', 'digex', 'com', 'digex', 'net', 'not', 'for', 'mail', 'From', 'prb', 'access', 'digex', 'net', 'Pat', 'Newsgroups', 'sci', 'space', 'Subject', 'Re', 'Space', 'Manuevering', 'Tug', 'was', 'HST', 'servicing', 'mission_', 'Date', '1', 'May', '1993', '15', '25', '38', '0400', 'Organization', 'Express', 'Access', 'Online', 'Communications', 'USA', 'Lines', '83', 'Message', 'ID', '1ruiri', 'dig', 'access', 'digex', 'net', 'References', '1rnaih', 'jvj', 'access', 'digex', 'net', 'C6BBow', 'IH9', 'news', 'cso', 'uiuc', 'edu', 'NNTP', 'Posting', 'Host', 'access', 'digex', 'net', 'In', 'article', 'C6BBow', 'IH9', 'news', 'cso', 'uiuc', 'edu', 'jbh55289', 'uxa', 'cso', 'uiuc', 'edu', 'Josh', 'Hopkins', 'writes', 'prb', 'access', 'digex', 'net', 'Pat', 'writes', 'Given', 'that', 'what', 'i', 'described', 'for', 'the', 'HST', 'seemed', 'to', 'be', 'the', 'SMT', 'and', 'given', 'the', 'mass', 'amrgins', 'on', 'the', 'discovery', 'mission', 'is', 'tight', 'enough', 'that', 'spacewalking', 'has', 'to', 'be', 'carefully', 'constrained', 'No', 'EDO', 'pallets', 'no', 'spare', 'Suits', 'no', 'extra', 'MMU', 's', 'Has', 'someone', 'actually', 'verified', 'that', 'mass', 'is', 'the', 'predominant', 'constraint', 'on', 'this', 'mission', 'You', 'seem', 'to', 'be', 'assuming', 'it', 'without', 'giving', 'supporting', 'evidence', 'Someone', 'from', 'NASA', 'posted', 'that', 'there', 'were', 'very', 'significant', 'mass', 'margins', 'on', 'the', 'HST', 're', 'boost', 'mission', 'A', 'while', 'back', 'i', 'had', 'asked', 'why', 'not', 'carry', 'the', 'EDO', 'pallet', 'up', 'and', 'the', 'answer', 'was', 'the', 'mass', 'margins', 'were', 'tight', 'enough', 'they', 'weren', 't', 'even', 'carrying', 'extra', 'suits', 'WHy', 'not', 'do', 'this', 'Quick', 'Test', 'Goldins', 'philosophjy', 'of', 'faster', 'cheaper', 'better', 'Build', 'a', 'real', 'fast', 'Space', 'TUg', 'to', 'handle', 'the', 're', 'boost', 'of', 'the', 'HST', 'using', 'clean', 'Cryo', 'fuels', 'and', 'get', 'it', 'ready', 'before', 'the', 'HST', 'mission', 'Pat', 'this', 'would', 'be', 'slower', 'more', 'expensive', 'and', 'worse', 'Slower', 'The', 'shuttle', 'mission', 'is', 'scheduled', 'to', 'go', 'up', 'in', 'December', 'That', 's', 'less', 'than', 'eight', 'months', 'away', 'There', 'is', 'no', 'way', 'you', 'could', 'build', 'new', 'hardware', 'retrain', 'and', 'reschedule', 'the', 'EVA', 's', 'in', 'that', 'time', 'Where', 's', 'wingo', 'when', 'you', 'need', 'him', 'COme', 'on', 'Knock', 'that', 'S', 'T', 'off', 'YOu', 'forget', 'that', 'during', 'skylab', 'they', 'did', 'overnight', 'mission', 'planning', 'for', 'the', 'repair', 'EVA', 's', 'Also', 'during', 'the', 'Intelsat', 'Mission', 'they', 'did', 'overnight', 'WETF', 'simulations', 'I', 'somehow', 'think', 'they', 'could', 'train', 'up', 'a', 'new', 'EVA', 'in', '8', 'months', 'And', 'as', 'for', 'building', 'hardware', 'anything', 'can', 'be', 'built', 'if', 'you', 'want', 'it', 'bad', 'enough', 'YOu', 'forget', 'the', 'BUS', '1', 'is', 'already', 'built', 'all', 'they', 'd', 'ahve', 'to', 'do', 'is', 'soup', 'it', 'up', 'even', 'test', 'it', 'on', 'a', 'delta', 'mission', 'Don', 't', 'get', 'into', 'this', 'mode', 'of', 'negativism', 'besides', 'at', 'the', 'rate', 'missions', 'slip', 'the', 'Discovery', 'won', 't', 'launch', 'on', 'this', 'mission', 'until', 'March', 'that', 's', 'almost', 'a', 'year', 'More', 'Expensive', 'Your', 'proposal', 'still', 'requires', 'the', 'shuttle', 'to', 'do', 'everything', 'it', 'was', 'going', 'to', 'do', 'execpt', 'fire', 'the', 'OMS', 'In', 'addition', 'you', 've', 'added', 'significant', 'extra', 'cost', 'for', 'a', 'new', 'piece', 'of', 'complex', 'hardware', 'Ah', 'but', 'how', 'much', 'more', 'expensive', 'is', 'the', 'Second', 'HST', 'servicing', 'mission', 'YOu', 'forget', 'there', 'is', 'a', 'bum', 'FGS', 'the', 'Solar', 'array', 'electronics', 'are', 'getting', 'hinky', 'and', 'there', 'is', 'still', '8', 'months', 'until', 'the', 'servicing', 'mission', 'The', 'time', 'for', 'the', 'space', 'walks', 'are', 'growing', 'rapidly', 'THis', 'was', 'orignally', 'planned', 'out', 'as', '3', 'spacewalks', 'now', 'they', 'are', 'at', '5', 'EVA', 's', 'with', '3', 'reserve', 'walks', 'If', 'the', 'SMT', 'can', 'avoid', 'a', 'second', 'servicing', 'mission', 'that', 's', '500', 'million', 'saved', 'If', 'the', 'Weight', 'savings', 'means', 'they', 'can', 'sit', 'on', 'orbit', 'for', '30', 'Days', 'and', 'handle', 'any', 'contingency', 'problems', 'that', 's', 'quite', 'a', 'savings', 'According', 'to', 'a', 'GAO', 'report', 'on', 'the', 'OMV', 'I', 'have', 'before', 'me', 'there', 'are', 'only', 'two', 'currently', 'planned', 'missions', 'that', 'could', 'use', 'such', 'a', 'vehicle', 'HST', 'and', 'AXAF', 'Since', 'AXAF', 'has', 'since', 'been', 'scaled', 'back', 'and', 'HST', 'can', 'rely', 'on', 'the', 'shuttle', 'there', 'doesn', 't', 'seem', 'to', 'be', 'any', 'need', 'for', 'your', 'vehicle', 'Of', 'course', 'there', 'wasn', 't', 'any', 'need', 'for', 'the', 'Saturn', 'V', 'after', 'apollo', 'too', 'as', 'for', 'the', 'problems', 'with', 'the', 'aperture', 'door', 'I', 'am', 'sure', 'they', 'can', 'work', 'out', 'some', 'way', 'to', 'handle', 'that', 'Maybe', 'a', 'Plug', 'made', 'from', 'Frozen', 'ice', 'it', 'll', 'keep', 'out', 'any', 'contamination', 'yet', 'sublime', 'away', 'after', 'teh', 'boost', 'pat']\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "yrlw1Gib3Coz",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 87
        },
        "outputId": "b9aab2d0-1601-4fbb-c2cd-4737b86817e6"
      },
      "source": [
        "# This is the equivalent place that the previous implementations \n",
        "# started\n",
        "document_token_counts = tokenized_texts.map(Counter)\n",
        "print('\\n\\ndocument-level counts of first document')\n",
        "print(document_token_counts.first().most_common(10))"
      ],
      "execution_count": 16,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "\n",
            "\n",
            "document-level counts of first document\n",
            "[('the', 31), ('that', 12), ('to', 11), ('mission', 11), ('a', 11), ('for', 10), ('and', 9), ('on', 9), ('is', 9), ('s', 9)]\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "eZEbtGxk3D96",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 87
        },
        "outputId": "141e63c1-e557-4f1f-83e5-5e7a7e8c89c2"
      },
      "source": [
        "word_counts = token_counts = document_token_counts.reduce(add)\n",
        "print('\\n\\nword counts')\n",
        "print(word_counts.most_common(10))"
      ],
      "execution_count": 17,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "\n",
            "\n",
            "word counts\n",
            "[('the', 1648), ('of', 804), ('edu', 784), ('to', 770), ('and', 641), ('a', 615), ('in', 402), ('is', 352), ('for', 305), ('cmu', 288)]\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "6cqJ7p9s3GPo",
        "colab_type": "text"
      },
      "source": [
        "As you see, we use the map and reduce methods here. Spark allows you to implement MapReduce-style programs, but you can also implement in many other ways."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "9Crtr94y3JTF",
        "colab_type": "text"
      },
      "source": [
        "#### Python and R\n",
        "\n",
        "Spark is primarily implemented in Scala. The Java API is there to allow more idiomatic Java use of Spark.     There is also a Python API (PySpark) and an R API (SparkR). Spark-based programs implemented in Scala or Java run on the same JVM that serves as the driver. Programs implemented in PySpark or SparkR run in Python and R processes, respectively, with the SparkSession ultimately in a different process. This generally does not affect the performance, unless we use functions defined in Python or R.\n",
        "\n",
        "As can be seen in the previous example, when we are tokenizing, counting, and combining counts we are calling Python code to process our data. This is accomplished by the JVM process serializing and shipping the data to the Python process, which is then deserialized, processed, serialized, and shipped back to the JVM to be deserialized. This adds a lot of extra work to our job. When using PySpark or SparkR it will be faster to use internal Spark functions whenever possible.\n",
        "\n",
        "Not using custom functions in Python or R seems restrictive when using `RDD`s, but most likely, your work will be using the `DataFrame`s and `DataSet`s that we discuss in the next section  .  "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "CjGyqDiA3Pzd",
        "colab_type": "text"
      },
      "source": [
        "## Spark SQL and Spark MLlib\n",
        "\n",
        "Since the release of Spark 2,  the primary intended way to work with data within Spark is through the `Dataset`. The `Dataset[T]` is an object that allows us to treat our distributed data as tables. The type parameter `T` is the type used to represent the rows of the table. There is a special kind of `Dataset` in which the type of the rows is `Row`, which allows us to have tabular data without defining new classes—this does come at the cost of losing some type safety. The examples we'll be using will generally be with `DataFrame`s, since they are the best way to work with data in PySpark.\n",
        "\n",
        "The `Dataset` and `DataFrame` are defined in the Spark SQL module, since one of the greatest benefits is the ability to express many operations with SQL.   The prebuilt user-defined functions (`UDF`s) are available in all the APIs. This allows us to do most kinds of processing in the non-JVM languages with the same efficiency as if we were using Scala or Java.\n",
        "\n",
        "Another module we need to introduce before we begin talking about Spark NLP is MLlib. MLlib is a module for doing machine learning on Spark. Before Spark 2, all the MLlib algorithms were implemented on `RDD`s. Since then, a new version of MLlib was defined using `Dataset`s and `DataFrame`s. MLlib is similar in design, at a high level, to other machine learning libraries, with a notion of transformers, models, and pipelines.\n",
        "\n",
        "Before we talk about MLlib, let's load some data into a `DataFrame`, since MLlib is built using `DataFrame`s.  We will be using the Iris data set, which is often used as an example in data science. It's small, easy to understand, and can work for clustering and classification examples. It is structured data, so it doesn't give us any text data to work with. Table-like structure is generally designed around structured data, so this data will help us explore the API before getting into using Spark for text.\n",
        "\n",
        "The `iris.data` file does not have a header, so we have to tell Spark what the columns are when they are loaded. Let's construct a schema.  The schema is the definition of the columns and their types in the `DataFrame`. The most common task is to build a model to predict what class an iris flower is (I. virginica, I. setosa, or I. versicolor) based on its sepals and petals."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "iarB5A523E_5",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.sql.types import *"
      ],
      "execution_count": 20,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "od1TGFPv36cS",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "schema = StructType([\n",
        "    StructField('sepal_length', DoubleType(), nullable=False),\n",
        "    StructField('sepal_width', DoubleType(), nullable=False),\n",
        "    StructField('petal_length', DoubleType(), nullable=False),\n",
        "    StructField('petal_width', DoubleType(), nullable=False),\n",
        "    StructField('class', StringType(), nullable=False)\n",
        "])"
      ],
      "execution_count": 26,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "uXkOH7Fv3_oJ",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "iris = spark.read.csv('./data/iris.data', schema=schema)\n"
      ],
      "execution_count": 27,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "k8mb_RIV4gG5",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "001fbfff-4432-44f3-e69c-fdc0e8186d91"
      },
      "source": [
        "iris.describe().toPandas()"
      ],
      "execution_count": 28,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>summary</th>\n",
              "      <th>sepal_length</th>\n",
              "      <th>sepal_width</th>\n",
              "      <th>petal_length</th>\n",
              "      <th>petal_width</th>\n",
              "      <th>class</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>count</td>\n",
              "      <td>150</td>\n",
              "      <td>150</td>\n",
              "      <td>150</td>\n",
              "      <td>150</td>\n",
              "      <td>150</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>mean</td>\n",
              "      <td>5.843333333333335</td>\n",
              "      <td>3.0540000000000007</td>\n",
              "      <td>3.7586666666666693</td>\n",
              "      <td>1.1986666666666672</td>\n",
              "      <td>None</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>stddev</td>\n",
              "      <td>0.8280661279778637</td>\n",
              "      <td>0.43359431136217375</td>\n",
              "      <td>1.764420419952262</td>\n",
              "      <td>0.7631607417008414</td>\n",
              "      <td>None</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>min</td>\n",
              "      <td>4.3</td>\n",
              "      <td>2.0</td>\n",
              "      <td>1.0</td>\n",
              "      <td>0.1</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>max</td>\n",
              "      <td>7.9</td>\n",
              "      <td>4.4</td>\n",
              "      <td>6.9</td>\n",
              "      <td>2.5</td>\n",
              "      <td>Iris-virginica</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "  summary        sepal_length  ...         petal_width           class\n",
              "0   count                 150  ...                 150             150\n",
              "1    mean   5.843333333333335  ...  1.1986666666666672            None\n",
              "2  stddev  0.8280661279778637  ...  0.7631607417008414            None\n",
              "3     min                 4.3  ...                 0.1     Iris-setosa\n",
              "4     max                 7.9  ...                 2.5  Iris-virginica\n",
              "\n",
              "[5 rows x 6 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 28
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "2F9XRpbd4p9w",
        "colab_type": "text"
      },
      "source": [
        "Let's start by looking at some of the summary stats for the Iris setosa class."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Xy9s9xWP4k9R",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 137
        },
        "outputId": "b2cfda21-4008-4d3d-9691-1e1686b79019"
      },
      "source": [
        "iris.select('class').distinct().toPandas()"
      ],
      "execution_count": 29,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>class</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>Iris-virginica</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>Iris-versicolor</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "             class\n",
              "0   Iris-virginica\n",
              "1      Iris-setosa\n",
              "2  Iris-versicolor"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 29
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "zbsRnoJe4qiv",
        "colab_type": "text"
      },
      "source": [
        "We can register a `DataFrame`, which will allow us to interact with it purely through SQL. We will be registering our `DataFrame` as a temporary table. This means that the table will exist only for the lifetime of our `App`, and it will be available only through our `App`'s `SparkSession`."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "zQeIiXFk4nov",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "iris.registerTempTable('iris')"
      ],
      "execution_count": 30,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "2Nr8T9Lb42Ng",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "dc6cfc0d-5f74-411b-94a0-2814722b80dd"
      },
      "source": [
        "spark.sql('''\n",
        "SELECT *\n",
        "FROM iris\n",
        "LIMIT 5\n",
        "''').toPandas()"
      ],
      "execution_count": 31,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>sepal_length</th>\n",
              "      <th>sepal_width</th>\n",
              "      <th>petal_length</th>\n",
              "      <th>petal_width</th>\n",
              "      <th>class</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>5.1</td>\n",
              "      <td>3.5</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>4.9</td>\n",
              "      <td>3.0</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>4.7</td>\n",
              "      <td>3.2</td>\n",
              "      <td>1.3</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>4.6</td>\n",
              "      <td>3.1</td>\n",
              "      <td>1.5</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>5.0</td>\n",
              "      <td>3.6</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "   sepal_length  sepal_width  petal_length  petal_width        class\n",
              "0           5.1          3.5           1.4          0.2  Iris-setosa\n",
              "1           4.9          3.0           1.4          0.2  Iris-setosa\n",
              "2           4.7          3.2           1.3          0.2  Iris-setosa\n",
              "3           4.6          3.1           1.5          0.2  Iris-setosa\n",
              "4           5.0          3.6           1.4          0.2  Iris-setosa"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 31
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "niPjvvGV45Qe",
        "colab_type": "text"
      },
      "source": [
        "Let's look at some of the fields grouped by their class."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Tc847MrT43HQ",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 137
        },
        "outputId": "837314c2-b010-447e-c0ef-24bdfc37caa5"
      },
      "source": [
        "spark.sql('''\n",
        "SELECT \n",
        "    class, \n",
        "    min(sepal_length), avg(sepal_length), max(sepal_length),\n",
        "    min(sepal_width), avg(sepal_width), max(sepal_width),\n",
        "    min(petal_length), avg(petal_length), max(petal_length),\n",
        "    min(petal_width), avg(petal_width), max(petal_width)\n",
        "FROM iris\n",
        "GROUP BY class\n",
        "''').toPandas()"
      ],
      "execution_count": 32,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>class</th>\n",
              "      <th>min(sepal_length)</th>\n",
              "      <th>avg(sepal_length)</th>\n",
              "      <th>max(sepal_length)</th>\n",
              "      <th>min(sepal_width)</th>\n",
              "      <th>avg(sepal_width)</th>\n",
              "      <th>max(sepal_width)</th>\n",
              "      <th>min(petal_length)</th>\n",
              "      <th>avg(petal_length)</th>\n",
              "      <th>max(petal_length)</th>\n",
              "      <th>min(petal_width)</th>\n",
              "      <th>avg(petal_width)</th>\n",
              "      <th>max(petal_width)</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>Iris-virginica</td>\n",
              "      <td>4.9</td>\n",
              "      <td>6.588</td>\n",
              "      <td>7.9</td>\n",
              "      <td>2.2</td>\n",
              "      <td>2.974</td>\n",
              "      <td>3.8</td>\n",
              "      <td>4.5</td>\n",
              "      <td>5.552</td>\n",
              "      <td>6.9</td>\n",
              "      <td>1.4</td>\n",
              "      <td>2.026</td>\n",
              "      <td>2.5</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>4.3</td>\n",
              "      <td>5.006</td>\n",
              "      <td>5.8</td>\n",
              "      <td>2.3</td>\n",
              "      <td>3.418</td>\n",
              "      <td>4.4</td>\n",
              "      <td>1.0</td>\n",
              "      <td>1.464</td>\n",
              "      <td>1.9</td>\n",
              "      <td>0.1</td>\n",
              "      <td>0.244</td>\n",
              "      <td>0.6</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>Iris-versicolor</td>\n",
              "      <td>4.9</td>\n",
              "      <td>5.936</td>\n",
              "      <td>7.0</td>\n",
              "      <td>2.0</td>\n",
              "      <td>2.770</td>\n",
              "      <td>3.4</td>\n",
              "      <td>3.0</td>\n",
              "      <td>4.260</td>\n",
              "      <td>5.1</td>\n",
              "      <td>1.0</td>\n",
              "      <td>1.326</td>\n",
              "      <td>1.8</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "             class  min(sepal_length)  ...  avg(petal_width)  max(petal_width)\n",
              "0   Iris-virginica                4.9  ...             2.026               2.5\n",
              "1      Iris-setosa                4.3  ...             0.244               0.6\n",
              "2  Iris-versicolor                4.9  ...             1.326               1.8\n",
              "\n",
              "[3 rows x 13 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 32
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "GMOw8zNL48P3",
        "colab_type": "text"
      },
      "source": [
        "### Transformers\n",
        "\n",
        "A `Transformer` is a piece of logic that transforms the data without needing to learn or fit anything from the data. A good way to understand transformers is that they represent functions that we wish to map over our data. All stages of a pipeline have parameters so that we can make sure that the transformation is being applied to the right fields and with the desired configuration. Let's look at a few examples.\n",
        "\n",
        "#### SQLTransformer\n",
        "\n",
        "The `SQLTransformer` has only one parameter—`statement`—which is the SQL statement that will be executed against our `DataFrame`. Let's use an `SQLTransformer` to do the group-by we performed previously. table0306 shows the result."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "kXQnJvqm46sY",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.feature import SQLTransformer\n",
        "\n",
        "statement = '''\n",
        "SELECT \n",
        "    class, \n",
        "    min(sepal_length), avg(sepal_length), max(sepal_length),\n",
        "    min(sepal_width), avg(sepal_width), max(sepal_width),\n",
        "    min(petal_length), avg(petal_length), max(petal_length),\n",
        "    min(petal_width), avg(petal_width), max(petal_width)\n",
        "FROM iris\n",
        "GROUP BY class\n",
        "'''\n",
        "\n",
        "sql_transformer = SQLTransformer(statement=statement)"
      ],
      "execution_count": 33,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "nnzBVd1w5Xw4",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 137
        },
        "outputId": "cd6543ec-ffea-4551-ed01-86b0978f2605"
      },
      "source": [
        "sql_transformer.transform(iris).toPandas()"
      ],
      "execution_count": 34,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>class</th>\n",
              "      <th>min(sepal_length)</th>\n",
              "      <th>avg(sepal_length)</th>\n",
              "      <th>max(sepal_length)</th>\n",
              "      <th>min(sepal_width)</th>\n",
              "      <th>avg(sepal_width)</th>\n",
              "      <th>max(sepal_width)</th>\n",
              "      <th>min(petal_length)</th>\n",
              "      <th>avg(petal_length)</th>\n",
              "      <th>max(petal_length)</th>\n",
              "      <th>min(petal_width)</th>\n",
              "      <th>avg(petal_width)</th>\n",
              "      <th>max(petal_width)</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>Iris-virginica</td>\n",
              "      <td>4.9</td>\n",
              "      <td>6.588</td>\n",
              "      <td>7.9</td>\n",
              "      <td>2.2</td>\n",
              "      <td>2.974</td>\n",
              "      <td>3.8</td>\n",
              "      <td>4.5</td>\n",
              "      <td>5.552</td>\n",
              "      <td>6.9</td>\n",
              "      <td>1.4</td>\n",
              "      <td>2.026</td>\n",
              "      <td>2.5</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>4.3</td>\n",
              "      <td>5.006</td>\n",
              "      <td>5.8</td>\n",
              "      <td>2.3</td>\n",
              "      <td>3.418</td>\n",
              "      <td>4.4</td>\n",
              "      <td>1.0</td>\n",
              "      <td>1.464</td>\n",
              "      <td>1.9</td>\n",
              "      <td>0.1</td>\n",
              "      <td>0.244</td>\n",
              "      <td>0.6</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>Iris-versicolor</td>\n",
              "      <td>4.9</td>\n",
              "      <td>5.936</td>\n",
              "      <td>7.0</td>\n",
              "      <td>2.0</td>\n",
              "      <td>2.770</td>\n",
              "      <td>3.4</td>\n",
              "      <td>3.0</td>\n",
              "      <td>4.260</td>\n",
              "      <td>5.1</td>\n",
              "      <td>1.0</td>\n",
              "      <td>1.326</td>\n",
              "      <td>1.8</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "             class  min(sepal_length)  ...  avg(petal_width)  max(petal_width)\n",
              "0   Iris-virginica                4.9  ...             2.026               2.5\n",
              "1      Iris-setosa                4.3  ...             0.244               0.6\n",
              "2  Iris-versicolor                4.9  ...             1.326               1.8\n",
              "\n",
              "[3 rows x 13 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 34
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cg8MfQqg5dyc",
        "colab_type": "text"
      },
      "source": [
        "We get the same output as when we ran the SQL command.\n",
        "\n",
        "SQLTransformer is useful when you have preprocessing or restructuring that you need to perform on your data before other steps in the pipeline. Now let's look at a transformer that works on one field and returns the original data with a new field."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0u134bux5fkc",
        "colab_type": "text"
      },
      "source": [
        "#### Binarizer\n",
        "\n",
        "The `Binarizer` is a `Transformer` that applies a threshold to a numeric field, turning it into 0s (when below the threshold) and 1s (when above the threshold). It takes three parameters:\n",
        "\n",
        "* inputCol  \n",
        "The column to be binarized\n",
        "* outputCol  \n",
        "The column containing the binarized values\n",
        "* threshold  \n",
        "The threshold we will apply"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "khg_A1gK5cbS",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.feature import Binarizer\n",
        "\n",
        "binarizer = Binarizer(\n",
        "    inputCol='sepal_length', \n",
        "    outputCol='sepal_length_above_5', \n",
        "    threshold=5.0\n",
        ")"
      ],
      "execution_count": 35,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "BBsbfx395v0A",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "d34c1e88-e0f5-4c84-e761-eb09558bcf14"
      },
      "source": [
        "binarizer.transform(iris).limit(5).toPandas()"
      ],
      "execution_count": 36,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>sepal_length</th>\n",
              "      <th>sepal_width</th>\n",
              "      <th>petal_length</th>\n",
              "      <th>petal_width</th>\n",
              "      <th>class</th>\n",
              "      <th>sepal_length_above_5</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>5.1</td>\n",
              "      <td>3.5</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>1.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>4.9</td>\n",
              "      <td>3.0</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>0.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>4.7</td>\n",
              "      <td>3.2</td>\n",
              "      <td>1.3</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>0.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>4.6</td>\n",
              "      <td>3.1</td>\n",
              "      <td>1.5</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>0.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>5.0</td>\n",
              "      <td>3.6</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>0.0</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "   sepal_length  sepal_width  ...        class  sepal_length_above_5\n",
              "0           5.1          3.5  ...  Iris-setosa                   1.0\n",
              "1           4.9          3.0  ...  Iris-setosa                   0.0\n",
              "2           4.7          3.2  ...  Iris-setosa                   0.0\n",
              "3           4.6          3.1  ...  Iris-setosa                   0.0\n",
              "4           5.0          3.6  ...  Iris-setosa                   0.0\n",
              "\n",
              "[5 rows x 6 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 36
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "gPilfKc85xsG",
        "colab_type": "text"
      },
      "source": [
        "Unlike the `SQLTransformer`, the `Binarizer` returns a modified version of the input `DataFrame`. Almost all Transformers behave this way.\n",
        "\n",
        "The `Binarizer` is used when you want to convert a real valued property into a class. For example, if we want to mark social media posts as \"viral\" and \"not-viral\" we could use a `Binarizer` on the views property."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "aXAFplFK53sH",
        "colab_type": "text"
      },
      "source": [
        "#### VectorAssembler\n",
        "\n",
        "Another import Transformer is the VectorAssembler. It takes a list of numeric and vector-valued columns and constructs a single vector. This is useful because all MLlib's machine learning algorithms expect a single vector-valued input column for features. The VectorAssembler takes two parameters:\n",
        "\n",
        "* inputCols  \n",
        "The list of columns to be assembled\n",
        "* outputCol  \n",
        "The column containing the new vectors"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "g70IjXs558yQ",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.feature import VectorAssembler\n",
        "\n",
        "assembler = VectorAssembler(\n",
        "    inputCols=[\n",
        "        'sepal_length', 'sepal_width', \n",
        "        'petal_length', 'petal_width'\n",
        "    ], \n",
        "    outputCol='features'\n",
        ")"
      ],
      "execution_count": 37,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "eTTAhTU35-rJ",
        "colab_type": "text"
      },
      "source": [
        "Let's persist this data"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Q4EPffJx5-QO",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "iris_w_vecs = assembler.transform(iris).persist()"
      ],
      "execution_count": 38,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "KqWV8YQF6A1A",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "01e4aadf-5dc0-4152-c7f2-8584660a65f2"
      },
      "source": [
        "iris_w_vecs.limit(5).toPandas()"
      ],
      "execution_count": 39,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>sepal_length</th>\n",
              "      <th>sepal_width</th>\n",
              "      <th>petal_length</th>\n",
              "      <th>petal_width</th>\n",
              "      <th>class</th>\n",
              "      <th>features</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>5.1</td>\n",
              "      <td>3.5</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[5.1, 3.5, 1.4, 0.2]</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>4.9</td>\n",
              "      <td>3.0</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.9, 3.0, 1.4, 0.2]</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>4.7</td>\n",
              "      <td>3.2</td>\n",
              "      <td>1.3</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.7, 3.2, 1.3, 0.2]</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>4.6</td>\n",
              "      <td>3.1</td>\n",
              "      <td>1.5</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.6, 3.1, 1.5, 0.2]</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>5.0</td>\n",
              "      <td>3.6</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[5.0, 3.6, 1.4, 0.2]</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "   sepal_length  sepal_width  ...        class              features\n",
              "0           5.1          3.5  ...  Iris-setosa  [5.1, 3.5, 1.4, 0.2]\n",
              "1           4.9          3.0  ...  Iris-setosa  [4.9, 3.0, 1.4, 0.2]\n",
              "2           4.7          3.2  ...  Iris-setosa  [4.7, 3.2, 1.3, 0.2]\n",
              "3           4.6          3.1  ...  Iris-setosa  [4.6, 3.1, 1.5, 0.2]\n",
              "4           5.0          3.6  ...  Iris-setosa  [5.0, 3.6, 1.4, 0.2]\n",
              "\n",
              "[5 rows x 6 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 39
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "HOAHYDh96DVb",
        "colab_type": "text"
      },
      "source": [
        "### Estimators and Models\n",
        "\n",
        "`Estimator`s allow us to create transformations that are informed by our data. Classification models (e.g., decision trees) and regression models (e.g., linear regressions) are prominent examples, but some preprocessing algorithms are like this as well. For example, preprocessing that needs to know the whole vocabulary first will be `Estimator`s. The `Estimator` is fit with a `DataFrame` and returns a `Model`, which is a kind of Transformer. The `Model`s created from classifier and regression `Estimator`s are `PredictionModel`s.\n",
        "\n",
        "This is a similar design to scikit-learn, with the exception that in scikit-learn when we call `fit` we mutate the estimator instead of creating a new object. There are pros and cons to this, as there always are when debating mutability. Idiomatic Scala strongly prefers immutability.\n",
        "\n",
        "Let's look at some examples of `Estimator`s and `Model`s.\n",
        "\n",
        "#### MinMaxScaler\n",
        "\n",
        "The MinMaxScaler allows us to scale our data to be between 0 and 1. It takes four parameters:\n",
        "\n",
        "* inputCol  \n",
        "The column to be scaled\n",
        "* outputCol  \n",
        "The column containing the scaled values\n",
        "* max  \n",
        "The new maximum value (optional, default = 1)\n",
        "* min  \n",
        "The new minimum value (optional, default = 0)"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Dc5dBhtZ6B4w",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.feature import MinMaxScaler\n",
        "\n",
        "scaler = MinMaxScaler(\n",
        "    inputCol='features', \n",
        "    outputCol='petal_length_scaled'\n",
        ")"
      ],
      "execution_count": 40,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "wepcwUUX6h0v",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "scaler_model = scaler.fit(iris_w_vecs)\n"
      ],
      "execution_count": 41,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "YqiwVXvr6iu_",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "ccfbd608-f1df-497c-df50-74d07dc44356"
      },
      "source": [
        "scaler_model.transform(iris_w_vecs).limit(5).toPandas()\n"
      ],
      "execution_count": 42,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>sepal_length</th>\n",
              "      <th>sepal_width</th>\n",
              "      <th>petal_length</th>\n",
              "      <th>petal_width</th>\n",
              "      <th>class</th>\n",
              "      <th>features</th>\n",
              "      <th>petal_length_scaled</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>5.1</td>\n",
              "      <td>3.5</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[5.1, 3.5, 1.4, 0.2]</td>\n",
              "      <td>[0.22222222222222213, 0.6249999999999999, 0.06...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>4.9</td>\n",
              "      <td>3.0</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.9, 3.0, 1.4, 0.2]</td>\n",
              "      <td>[0.1666666666666668, 0.41666666666666663, 0.06...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>4.7</td>\n",
              "      <td>3.2</td>\n",
              "      <td>1.3</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.7, 3.2, 1.3, 0.2]</td>\n",
              "      <td>[0.11111111111111119, 0.5, 0.05084745762711865...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>4.6</td>\n",
              "      <td>3.1</td>\n",
              "      <td>1.5</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.6, 3.1, 1.5, 0.2]</td>\n",
              "      <td>[0.08333333333333327, 0.4583333333333333, 0.08...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>5.0</td>\n",
              "      <td>3.6</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[5.0, 3.6, 1.4, 0.2]</td>\n",
              "      <td>[0.19444444444444448, 0.6666666666666666, 0.06...</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "   sepal_length  ...                                petal_length_scaled\n",
              "0           5.1  ...  [0.22222222222222213, 0.6249999999999999, 0.06...\n",
              "1           4.9  ...  [0.1666666666666668, 0.41666666666666663, 0.06...\n",
              "2           4.7  ...  [0.11111111111111119, 0.5, 0.05084745762711865...\n",
              "3           4.6  ...  [0.08333333333333327, 0.4583333333333333, 0.08...\n",
              "4           5.0  ...  [0.19444444444444448, 0.6666666666666666, 0.06...\n",
              "\n",
              "[5 rows x 7 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 42
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "pCn3_GG76lmz",
        "colab_type": "text"
      },
      "source": [
        "Notice that the `petal_length_scaled` column now has values between 0 and 1. This can help some training algorithms, specifically those that have difficulty combining features of different scales."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "QsGomFZx6n-P",
        "colab_type": "text"
      },
      "source": [
        "#### StringIndexer\n",
        "\n",
        "Let's build a model! We will try and predict the class from the other features, and we will use a decision tree. First, though, we must convert our target into index values.\n",
        "\n",
        "The `StringIndexer` `Estimator` will turn our class values into indices. We want to do this to simplify some of the downstream processing. It is simpler to implement most training algorithms with the assumption that the target is a number. The `StringIndexer` takes four parameters:\n",
        "\n",
        "* inputCol  \n",
        "The column to be indexed\n",
        "* outputCol  \n",
        "The column containing the indexed values\n",
        "* handleInvalid  \n",
        "The policy for how the model should handle values not seen by the estimator (optional, default = error)\n",
        "* stringOrderType  \n",
        "How to order the values to make the indexing deterministic (optional, default = frequencyDesc)\n",
        "\n",
        "We will also want an `IndexToString` `Transformer`. This will let us map our predictions, which will be indices, back to string values. IndexToString takes three parameters:\n",
        "\n",
        "* inputCol  \n",
        "The column to be mapped\n",
        "* outputCol  \n",
        "The column containing the mapped values\n",
        "* labels  \n",
        "The mapping from index to value, usually generated by StringIndexer"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "pXK1Egxc6kAO",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.feature import StringIndexer, IndexToString\n",
        "\n",
        "indexer = StringIndexer(inputCol='class', outputCol='class_ix')\n",
        "indexer_model = indexer.fit(iris_w_vecs)\n",
        "\n",
        "index2string = IndexToString(\n",
        "    inputCol=indexer_model.getOrDefault('outputCol'), \n",
        "    outputCol='pred_class', \n",
        "    labels=indexer_model.labels\n",
        ")"
      ],
      "execution_count": 43,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "dNd2mh-S679n",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "iris_indexed = indexer_model.transform(iris_w_vecs)"
      ],
      "execution_count": 44,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "b3Fh42X36-O7",
        "colab_type": "text"
      },
      "source": [
        "Now we are ready to train our `DecisionTreeClassifier`. This `Estimator` has many parameters, so I recommend you become familiar with the APIs. They are all well documented in the [PySpark API documentation](https://spark.apache.org/docs/latest/api/python/index.html)."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "nQL3f3oc6873",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.classification import DecisionTreeClassifier\n",
        "\n",
        "dt_clfr = DecisionTreeClassifier(\n",
        "    featuresCol='features',\n",
        "    labelCol='class_ix',\n",
        "    maxDepth=5,\n",
        "    impurity='gini',\n",
        "    seed=123\n",
        ")"
      ],
      "execution_count": 45,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "1rmJmMg77OPX",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "dt_clfr_model = dt_clfr.fit(iris_indexed)"
      ],
      "execution_count": 46,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "4oVTGw9E7PGn",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "iris_w_pred = dt_clfr_model.transform(iris_indexed)"
      ],
      "execution_count": 47,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Lx6b899H7RoH",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "aad3013a-9879-4d1e-9cef-1f58d552e44d"
      },
      "source": [
        "iris_w_pred.limit(5).toPandas()"
      ],
      "execution_count": 48,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>sepal_length</th>\n",
              "      <th>sepal_width</th>\n",
              "      <th>petal_length</th>\n",
              "      <th>petal_width</th>\n",
              "      <th>class</th>\n",
              "      <th>features</th>\n",
              "      <th>class_ix</th>\n",
              "      <th>rawPrediction</th>\n",
              "      <th>probability</th>\n",
              "      <th>prediction</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>5.1</td>\n",
              "      <td>3.5</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[5.1, 3.5, 1.4, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>4.9</td>\n",
              "      <td>3.0</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.9, 3.0, 1.4, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>4.7</td>\n",
              "      <td>3.2</td>\n",
              "      <td>1.3</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.7, 3.2, 1.3, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>4.6</td>\n",
              "      <td>3.1</td>\n",
              "      <td>1.5</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.6, 3.1, 1.5, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>5.0</td>\n",
              "      <td>3.6</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[5.0, 3.6, 1.4, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "   sepal_length  sepal_width  ...      probability  prediction\n",
              "0           5.1          3.5  ...  [1.0, 0.0, 0.0]         0.0\n",
              "1           4.9          3.0  ...  [1.0, 0.0, 0.0]         0.0\n",
              "2           4.7          3.2  ...  [1.0, 0.0, 0.0]         0.0\n",
              "3           4.6          3.1  ...  [1.0, 0.0, 0.0]         0.0\n",
              "4           5.0          3.6  ...  [1.0, 0.0, 0.0]         0.0\n",
              "\n",
              "[5 rows x 10 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 48
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "O2zQHR8i7UB-",
        "colab_type": "text"
      },
      "source": [
        "Now we need to map the predicted classes back to their string form using our `IndexToString`."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "iqnw--uk7S0P",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "iris_w_pred_class = index2string.transform(iris_w_pred)"
      ],
      "execution_count": 49,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "2l40t2Vl7X31",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "5f3fa9da-b01e-48ca-e5e5-60ce9868fabe"
      },
      "source": [
        "iris_w_pred_class.limit(5).toPandas()"
      ],
      "execution_count": 50,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>sepal_length</th>\n",
              "      <th>sepal_width</th>\n",
              "      <th>petal_length</th>\n",
              "      <th>petal_width</th>\n",
              "      <th>class</th>\n",
              "      <th>features</th>\n",
              "      <th>class_ix</th>\n",
              "      <th>rawPrediction</th>\n",
              "      <th>probability</th>\n",
              "      <th>prediction</th>\n",
              "      <th>pred_class</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>5.1</td>\n",
              "      <td>3.5</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[5.1, 3.5, 1.4, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>4.9</td>\n",
              "      <td>3.0</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.9, 3.0, 1.4, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>4.7</td>\n",
              "      <td>3.2</td>\n",
              "      <td>1.3</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.7, 3.2, 1.3, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>4.6</td>\n",
              "      <td>3.1</td>\n",
              "      <td>1.5</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.6, 3.1, 1.5, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>5.0</td>\n",
              "      <td>3.6</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[5.0, 3.6, 1.4, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "   sepal_length  sepal_width  ...  prediction   pred_class\n",
              "0           5.1          3.5  ...         0.0  Iris-setosa\n",
              "1           4.9          3.0  ...         0.0  Iris-setosa\n",
              "2           4.7          3.2  ...         0.0  Iris-setosa\n",
              "3           4.6          3.1  ...         0.0  Iris-setosa\n",
              "4           5.0          3.6  ...         0.0  Iris-setosa\n",
              "\n",
              "[5 rows x 11 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 50
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "9M6aBVpm7Z-N",
        "colab_type": "text"
      },
      "source": [
        "How well did our model fit the data? Let's see how many predictions match the true class."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "v_3ruUra7d5y",
        "colab_type": "text"
      },
      "source": [
        "### Evaluators\n",
        "\n",
        "The evaluation options in MLlib are still limited compared to libraries like scikit-learn, but they can be useful if you are looking to create an easy-to-run training pipeline that calculates metrics.\n",
        "\n",
        "In our example, we are trying to solve a multiclass prediction problem, so we will use the `MulticlassClassificationEvaluator`."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "pDefCgpo7Y2e",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
        "\n",
        "evaluator = MulticlassClassificationEvaluator(\n",
        "    labelCol='class_ix', \n",
        "    metricName='accuracy'\n",
        ")"
      ],
      "execution_count": 51,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "N-1SaVgK7jmW",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 34
        },
        "outputId": "2f2b03b8-6655-4bcb-bf2b-10897a41c688"
      },
      "source": [
        "evaluator.evaluate(iris_w_pred_class)"
      ],
      "execution_count": 52,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "1.0"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 52
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "xFE0inS57mxw",
        "colab_type": "text"
      },
      "source": [
        "This seems too good. What if we are overfit? Perhaps we should try using cross-validation to evaluate our models. Before we do that, let's organize stages into a pipeline."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ng0U11uo7qZy",
        "colab_type": "text"
      },
      "source": [
        "### Pipelines\n",
        "\n",
        "Pipelines are a special kind of Estimator that takes a list of Transformers and Estimators and allows us to use them as a single Estimator "
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "BTlhmX0Y7lS3",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml import Pipeline\n",
        "\n",
        "pipeline = Pipeline(\n",
        "    stages=[assembler, indexer, dt_clfr, index2string]\n",
        ")"
      ],
      "execution_count": 53,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "3CYSiXjY7u12",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "pipeline_model = pipeline.fit(iris)"
      ],
      "execution_count": 54,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "SlQuCNcG7wGe",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "b64ca260-630d-4ffc-8758-38898628bd7f"
      },
      "source": [
        "pipeline_model.transform(iris).limit(5).toPandas()"
      ],
      "execution_count": 55,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>sepal_length</th>\n",
              "      <th>sepal_width</th>\n",
              "      <th>petal_length</th>\n",
              "      <th>petal_width</th>\n",
              "      <th>class</th>\n",
              "      <th>features</th>\n",
              "      <th>class_ix</th>\n",
              "      <th>rawPrediction</th>\n",
              "      <th>probability</th>\n",
              "      <th>prediction</th>\n",
              "      <th>pred_class</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>5.1</td>\n",
              "      <td>3.5</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[5.1, 3.5, 1.4, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>4.9</td>\n",
              "      <td>3.0</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.9, 3.0, 1.4, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>4.7</td>\n",
              "      <td>3.2</td>\n",
              "      <td>1.3</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.7, 3.2, 1.3, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>4.6</td>\n",
              "      <td>3.1</td>\n",
              "      <td>1.5</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[4.6, 3.1, 1.5, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>5.0</td>\n",
              "      <td>3.6</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0.2</td>\n",
              "      <td>Iris-setosa</td>\n",
              "      <td>[5.0, 3.6, 1.4, 0.2]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>[50.0, 0.0, 0.0]</td>\n",
              "      <td>[1.0, 0.0, 0.0]</td>\n",
              "      <td>0.0</td>\n",
              "      <td>Iris-setosa</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "   sepal_length  sepal_width  ...  prediction   pred_class\n",
              "0           5.1          3.5  ...         0.0  Iris-setosa\n",
              "1           4.9          3.0  ...         0.0  Iris-setosa\n",
              "2           4.7          3.2  ...         0.0  Iris-setosa\n",
              "3           4.6          3.1  ...         0.0  Iris-setosa\n",
              "4           5.0          3.6  ...         0.0  Iris-setosa\n",
              "\n",
              "[5 rows x 11 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 55
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "E_f-wAMr7yoI",
        "colab_type": "text"
      },
      "source": [
        "#### Cross validation\n",
        "Now that we have a `Pipeline` and an `Evaluator` we can create a `CrossValidator`. The `CrossValidator` itself is also an `Estimator`. When we call fit, it will fit our `pipeline` to each fold of data, and calculate the metric determined by our `Evaluator`. `CrossValidator` takes five parameters:\n",
        "\n",
        "* estimator  \n",
        "The `Estimator` to be tuned\n",
        "* estimatorParamMaps  \n",
        "The hyperparameter values to try in a hyperparameter grid search\n",
        "* evaluator  \n",
        "The `Evaluator` that calculates the metric\n",
        "* numFolds  \n",
        "The number of folds to split the data into\n",
        "* seed  \n",
        "A seed for making the splits reproducible\n",
        "\n",
        "We will make a trivial hyperparameter grid here, since we are only interested in estimating how well our model does on data it has not seen."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GdXmuDHj7w-P",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder\n",
        "\n",
        "param_grid = ParamGridBuilder().\\\n",
        "    addGrid(dt_clfr.maxDepth, [5]).\\\n",
        "    build()\n",
        "cv = CrossValidator(\n",
        "    estimator=pipeline, \n",
        "    estimatorParamMaps=param_grid,\n",
        "    evaluator=evaluator, \n",
        "    numFolds=3, \n",
        "    seed=123\n",
        ")"
      ],
      "execution_count": 56,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "9SHltwkM8CqZ",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "cv_model = cv.fit(iris)"
      ],
      "execution_count": 57,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "FUZs-fHz8EvK",
        "colab_type": "text"
      },
      "source": [
        "Now, we can see how the model does when trained on two-thirds and evaluated on one-third. The `avgMetrics` in `cv_model` contains the average value of the designated metric across folds for each point in the hyperparameter grid tested. In our case, there is only one point in the grid."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "E8bipoJ78Dyu",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 34
        },
        "outputId": "6d363955-e70f-4374-c758-bf84493c5545"
      },
      "source": [
        "cv_model.avgMetrics"
      ],
      "execution_count": 58,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[0.9588996659642801]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 58
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "vTDKEZ3D8Jve",
        "colab_type": "text"
      },
      "source": [
        ""
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "a4g-ER3_8NUw",
        "colab_type": "text"
      },
      "source": [
        "Keep in mind that 95% accuracy is much more believable than 100%.\n",
        "\n",
        "There are many other `Transformer`s, `Estimator`s, and `Model`s. We will look into more as we continue, but for now, there is one more thing we need to discuss—saving our pipelines. \n",
        "\n",
        "#### Serialization of models\n",
        "\n",
        "MLlib allows us to save `Pipelines` so that we can use them later. We can also save individual `Transformers` and `Models`, but we will often want to keep all the stages of a `Pipeline` together. Generally speaking, we use separate programs for building models and using models."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "u3KIXD5g8IPw",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "pipeline_model.write().overwrite().save('pipeline.model')"
      ],
      "execution_count": 59,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "L-RrvPhE8aj1",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 123
        },
        "outputId": "db996e65-d805-4622-d153-b379e0353b5e"
      },
      "source": [
        "! ls pipeline.model/*"
      ],
      "execution_count": 60,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "pipeline.model/metadata:\n",
            "part-00000  _SUCCESS\n",
            "\n",
            "pipeline.model/stages:\n",
            "0_VectorAssembler_fb9ec58ae5fa\t2_DecisionTreeClassifier_f3dd63f71dec\n",
            "1_StringIndexer_446dae549e67\t3_IndexToString_d09b78e10c1a\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "CDL8Agwr8dbb",
        "colab_type": "text"
      },
      "source": [
        "## NLP Libraries\n",
        "\n",
        "There are two kinds of NLP libraries, generally speaking: functionality libraries and annotation libraries.\n",
        "\n",
        "### Functionality Libraries\n",
        "\n",
        "A functionality library is a collection of functions built for specific NLP tasks and techniques. Often, the functions are built without assuming that other functions will be used first. This means that functions like part-of-speech (POS) tagging will also perform tokenization. These libraries are good for research because it is often much easier to implement novel functions. On the other hand, because there is no unifying design, the performance of these libraries is generally much worse than that of annotation libraries.\n",
        "\n",
        "The Natural Language Tool Kit (NLTK) is a great functionality library.  It was originally created by Edward Loper. The landmark NLP book Natural Language Processing with Python (O'Reilly) was written by Steven Bird, Ewan Klein, and Edward Loper. I strongly recommend that book to anyone learning NLP. There a many useful and interesting modules in NLTK. It is, and will likely remain, the best NLP library for teaching NLP. The functions are not necessarily implemented with runtime performance or other productionization concerns in mind. If you are working on a research project and using a data set manageable on a single machine, you should consider NLTK."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "nCQSbWMt8knB",
        "colab_type": "text"
      },
      "source": [
        "### Annotation Libraries\n",
        "   Annotation libraries are libraries in which all the functionality is built around a document-annotation model. There are three objects to keep in mind with annotation libraries: document, annotation, and annotator. The idea behind annotation libraries is to augment the incoming data with the results of our NLP functions.\n",
        "\n",
        "* Document  \n",
        "The document is the representation of the piece of text we wish to process. Naturally, the document must contain the text. Additionally, we often want to have an identifier associated with each document so that we can store our augmented data as structured data. This identifier will often be a title if the texts we are processing have titles.\n",
        "* Annotation  \n",
        "The annotation is the representation of the output of our NLP functions. For the annotation we need to have a type so that later processing knows how to interpret the annotations. Annotations also need to store their location within the document. For example, let's say the word \"pacing\" occurs 134 characters into the document. It will have 134 as the start, and 140 as the end. The lemma annotation for \"pacing\" will have the location. Some annotation libraries also have a concept of document-level annotation that does not have a location. There will be additional fields, depending on the type. Simple annotations like tokens generally don't have extra fields. Stem annotations usually have the stem that was extracted for the range of the text.\n",
        "* Annotator  \n",
        "The annotator is the object that contains the logic for using the NLP function. The annotator will often require configuration or external data sets. Additionally, there are model-based annotators. One of the benefits of an annotation library is that annotators can take advantage of the work done by previous annotators. This naturally creates a notion of a pipeline of annotators.\n",
        "\n",
        "#### spaCy\n",
        "\n",
        "spaCy is an \"industrial strength\" NLP library. I will give a brief description, but I encourage you to go and read their fantastic documentation. spaCy combines the document model just described with a model for the language being processed (English, Spanish, etc.), which has allowed spaCy to support multiple languages in a way that is easy for developers to use. Much of its functionality is implemented in Python to get the speed of native code. If you are working in an environment that is using only Python, and you are unlikely to run distributed processes, then spaCy is a great choice."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "gMV8HhcT8zcD",
        "colab_type": "text"
      },
      "source": [
        "### NLP in Other Libraries\n",
        "\n",
        "There are some non-NLP libraries that have some NLP functionality. It is often in machine learning libraries to support machine learning on text data.\n",
        "\n",
        "* scikit-learn  \n",
        "A Python machine learning library that has functionality for extracting features from text. This functionality is generally of the bag-of-words kind of processing. The way these processes are built allows them to easily take advantage of more NLP-focused libraries.\n",
        "* Lucene  \n",
        "A Java document search framework that has some text-processing functionality necessary for building a search engine. We will use Lucene later on when we talk about information retrieval.\n",
        "* Gensim  \n",
        "A topic-modeling library (and it performs other distributional semantics techniques). Like spaCy, it is partially implemented in Cython, and like scikit-learn, it allows plug-and-play text processing in its API."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "m5-K0vny88Ex",
        "colab_type": "text"
      },
      "source": [
        "## Spark NLP\n",
        "\n",
        "The Spark NLP library was originally designed in early 2017 as an annotation library native to Spark to take full advantage of Spark SQL and MLlib modules. The inspiration came from trying to use Spark to distribute other NLP libraries, which were generally not implemented with concurrency or distributed computing in mind.\n",
        "\n",
        "### Annotation Library\n",
        " Spark NLP has the same concepts as any other annotation library but differs in how it stores annotations. Most annotation libraries store the annotations in the document object, but Spark NLP creates columns for the different types of annotations.\n",
        "\n",
        "The annotators are implemented as `Transformers`, `Estimators`, and `Models`. Let's take a look at some examples.\n",
        "\n",
        "### Stages\n",
        "\n",
        "One of the design principles of Spark NLP is easy interoperability with the existing algorithms in MLlib. Because there is no notion of documents or annotations in MLlib there are transformers for turning text columns into documents and converting annotations into vanilla Spark SQL data types. The usual usage pattern is as follows:\n",
        "\n",
        "1. Load data with Spark SQL.\n",
        "2. Create document column.\n",
        "3. Process with Spark NLP.\n",
        "4. Convert annotations of interest into Spark SQL data types.\n",
        "5. Run additional MLlib stages.\n",
        "\n",
        "We have already looked at how to load data with Spark SQL and how to use MLlib stages in the standard Spark library, so we will look at the middle three stages now. First, we will look at the DocumentAssembler (stage 2)."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "FPIftf4H9NYN",
        "colab_type": "text"
      },
      "source": [
        "#### Transformers\n",
        "\n",
        "To explore these five stages  we will again use the mini_newsgroups data set"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "yocWLnFr8bkc",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from sparknlp import DocumentAssembler, Finisher"
      ],
      "execution_count": 61,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "EUiwFbpl9SHO",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# RDD containing filepath-text pairs\n",
        "texts = spark.sparkContext.wholeTextFiles(path)\n",
        "\n",
        "schema = StructType([\n",
        "    StructField('path', StringType()),\n",
        "    StructField('text', StringType()),\n",
        "])\n",
        "\n",
        "texts = spark.createDataFrame(texts, schema=schema)"
      ],
      "execution_count": 62,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "a7sR_06c9TSF",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "48a454d6-03f9-4e42-b4f4-c30a84b3b1aa"
      },
      "source": [
        "texts.limit(5).toPandas()"
      ],
      "execution_count": 63,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>path</th>\n",
              "      <th>text</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!das-news.harva...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Newsgroups: sci.space\\nPath: cantaloupe.srv.cs...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!magnesium.club...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Xref: cantaloupe.srv.cs.cmu.edu sci.space:6094...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!rochester!udel...</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "                                                path                                               text\n",
              "0  file:/content/data/mini_newsgroups/sci.space/6...  Path: cantaloupe.srv.cs.cmu.edu!das-news.harva...\n",
              "1  file:/content/data/mini_newsgroups/sci.space/6...  Newsgroups: sci.space\\nPath: cantaloupe.srv.cs...\n",
              "2  file:/content/data/mini_newsgroups/sci.space/6...  Path: cantaloupe.srv.cs.cmu.edu!magnesium.club...\n",
              "3  file:/content/data/mini_newsgroups/sci.space/6...  Xref: cantaloupe.srv.cs.cmu.edu sci.space:6094...\n",
              "4  file:/content/data/mini_newsgroups/sci.space/6...  Path: cantaloupe.srv.cs.cmu.edu!rochester!udel..."
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 63
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yPPQpNsV9Vpu",
        "colab_type": "text"
      },
      "source": [
        "#### DocumentAssembler\n",
        "\n",
        "The DocumentAssembler takes five parameters (see table0314):\n",
        "\n",
        "* inputCol  \n",
        "The column containing the text of the document\n",
        "* outputCol  \n",
        "The name of the column containing the newly constructed document\n",
        "* idCol  \n",
        "The name of the column containing the identifier (optional)\n",
        "* metadataCol  \n",
        "The name of a Map-type column that represents document metadata (optional)\n",
        "* trimAndClearNewLines  \n",
        "Determines whether to remove new line characters and trim strings (optional, default = True)"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "0Ek4VTEG9UQF",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "document_assembler = DocumentAssembler()\\\n",
        "    .setInputCol('text')\\\n",
        "    .setOutputCol('document')\\\n",
        "    .setIdCol('path')"
      ],
      "execution_count": 64,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "qaOFSlsT9etd",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "docs = document_assembler.transform(texts)"
      ],
      "execution_count": 65,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GsFEAbk19gLd",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "5fdfb6c0-0838-4bc6-801d-8e0ac39b9493"
      },
      "source": [
        "docs.limit(5).toPandas()"
      ],
      "execution_count": 66,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>path</th>\n",
              "      <th>text</th>\n",
              "      <th>document</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!das-news.harva...</td>\n",
              "      <td>[(document, 0, 3758, Path: cantaloupe.srv.cs.c...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Newsgroups: sci.space\\nPath: cantaloupe.srv.cs...</td>\n",
              "      <td>[(document, 0, 1579, Newsgroups: sci.space\\nPa...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!magnesium.club...</td>\n",
              "      <td>[(document, 0, 4409, Path: cantaloupe.srv.cs.c...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Xref: cantaloupe.srv.cs.cmu.edu sci.space:6094...</td>\n",
              "      <td>[(document, 0, 2248, Xref: cantaloupe.srv.cs.c...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!rochester!udel...</td>\n",
              "      <td>[(document, 0, 851, Path: cantaloupe.srv.cs.cm...</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "                                                path  ...                                           document\n",
              "0  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(document, 0, 3758, Path: cantaloupe.srv.cs.c...\n",
              "1  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(document, 0, 1579, Newsgroups: sci.space\\nPa...\n",
              "2  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(document, 0, 4409, Path: cantaloupe.srv.cs.c...\n",
              "3  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(document, 0, 2248, Xref: cantaloupe.srv.cs.c...\n",
              "4  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(document, 0, 851, Path: cantaloupe.srv.cs.cm...\n",
              "\n",
              "[5 rows x 3 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 66
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "MF6edG1V9hP1",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 161
        },
        "outputId": "d31cb75f-f6eb-4833-c2af-e910bd669ae3"
      },
      "source": [
        "docs.first()['document'][0].asDict()"
      ],
      "execution_count": 67,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "{'annotatorType': 'document',\n",
              " 'begin': 0,\n",
              " 'embeddings': [],\n",
              " 'end': 3758,\n",
              " 'metadata': {'id': 'file:/content/data/mini_newsgroups/sci.space/61534',\n",
              "  'sentence': '0'},\n",
              " 'result': \"Path: cantaloupe.srv.cs.cmu.edu!das-news.harvard.edu!noc.near.net!uunet!digex.com!digex.net!not-for-mail\\nFrom: prb@access.digex.net (Pat)\\nNewsgroups: sci.space\\nSubject: Re: Space Manuevering Tug (was HST servicing mission_)\\nDate: 1 May 1993 15:25:38 -0400\\nOrganization: Express Access Online Communications USA\\nLines: 83\\nMessage-ID: <1ruiri$dig@access.digex.net>\\nReferences: <1rnaih$jvj@access.digex.net> <C6BBow.IH9@news.cso.uiuc.edu>\\nNNTP-Posting-Host: access.digex.net\\n\\nIn article <C6BBow.IH9@news.cso.uiuc.edu> jbh55289@uxa.cso.uiuc.edu (Josh Hopkins) writes:\\n>prb@access.digex.net (Pat) writes:\\n|\\n|>Given that what i described for the HST  seemed to be the SMT,   and given\\n|>the mass amrgins on the discovery mission  is tight enough that  spacewalking\\n|>has to be carefully constrained.....  No EDO pallets,  no spare Suits,\\n|>no extra MMU's.   \\n|\\n|Has someone actually verified that mass is the predominant constraint on this\\n|mission?  You seem to be assuming it without giving supporting evidence.  \\n|\\n\\nSomeone from NASA posted that there were very significant mass margins\\non the HST re-boost mission.  A while back i had asked why not carry\\nthe EDO pallet up,  and the answer was the mass margins were tight enough, they weren't even carrying extra suits.\\n\\n|>WHy not do this?\\n|\\n|>\\tQuick Test  Goldins philosophjy  of faster cheaper, better.\\n|\\n|>Build a real fast Space TUg,  to handle the re-boost  of the HST  using\\n|>clean Cryo fuels,  and get it ready before the  HST mission.\\n|\\n|Pat, this would be slower, more expensive and worse.  \\n|\\n|Slower:  The shuttle mission is scheduled to go up in December.  That's less\\n|than eight months away.  There is no way you could build new hardware, retrain\\n|and reschedule the EVA's in that time.\\n|\\n\\nWhere's wingo when you need him:-)\\n\\nCOme on.   Knock that S**T off.\\n\\nYOu forget,  that during skylab,  they did  overnight mission planning\\nfor the repair EVA's.   Also during the��   \\nIntelsat Mission,   they did overnight  WETF simulations.\\nI somehow think they could train up a new  EVA in  8 months.\\n\\nAnd as for building hardware,  anything can  be built if you want it\\nbad enough.\\n\\nYOu forget,  the  BUS 1  is already built.  all they'd ahve to do\\nis soup it up, even test it  on a delta mission.\\n\\nDon't get into this mode of  negativism.   besides,  at the rate\\nmissions slip,   the Discovery won't launch on this mission until\\nMarch.  that's almost a year.\\n\\n|More Expensive:  Your proposal still requires the shuttle to do everything it\\n|was going to do execpt fire the OMS.  In addition, you've added significant\\n|extra cost for a new piece of complex hardware.\\n|\\n\\nAh,  but how much more expensive is the Second HST servicing mission.\\n\\nYOu forget,  there is a bum FGS,  the Solar array electronics, are\\ngetting hinky  and there is still 8 months until the servicing mission.\\n\\nThe time for the space walks are growing rapidly.  THis was orignally\\nplanned out as 3 spacewalks,  now they are at 5 EVA's  with 3 reserve\\nwalks.\\n\\nIf the SMT can avoid a second servicing mission that's $500 million\\nsaved.  If the Weight savings,  means they  can sit on orbit  for 30 Days.\\nand  handle any contingency  problems,  that's quite a savings.\\n\\n|\\n|According to a GAO report on the OMV I have before me, there are\\n|only two currently planned missions that could use such a vehicle -- HST and\\n|AXAF.  Since AXAF has since been scaled back and HST can rely on the shuttle,\\n|there doesn't seem to be any need for your vehicle.\\n\\n\\nOf course,  there wasn't any need for the Saturn  V  after apollo too.\\n\\nas for the problems with the aperture door,  I am sure they can\\nwork out some way to handle  that.  Maybe a Plug  made from\\nFrozen ice.?   it'll keep out any contamination,\\nyet sublime away  after teh boost.\\n\\npat\\n\\n\"}"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 67
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "X8jIEJFf9kvh",
        "colab_type": "text"
      },
      "source": [
        "### Annotators\n",
        "\n",
        "Now we look at stage 3—the annotators. This is the heart of the NLP work. So let's look at some of the Annotators available in Spark NLP.\n",
        "\n",
        "We will look at some commonly used annotators:\n",
        "\n",
        "* `SentenceDetector`\n",
        "* `Tokenizer`\n",
        "* `Lemmatizer`\n",
        "* `PerceptronApproach` (POSTagger)\n",
        "\n",
        "#### SentenceDetector\n",
        "\n",
        "The SentenceDetector uses a rule-based algorithm inspired by Kevin Dias's Ruby implementation. It takes the following parameters\n",
        "\n",
        "* inputCols  \n",
        "A list of columns to sentence-tokenize.\n",
        "* outputCol  \n",
        "The name of the new sentence column.\n",
        "* useAbbrevations  \n",
        "Determines whether to apply abbreviations at sentence detection.\n",
        "* useCustomBoundsOnly  \n",
        "Determines whether to only utilize custom bounds for sentence detection.\n",
        "* explodeSentences  \n",
        "Determines whether to explode each sentence into a different row, for better parallelization. Defaults to false.\n",
        "* customBounds  \n",
        "Characters used to explicitly mark sentence bounds."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ZWKNV7bV9ixl",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from sparknlp.annotator import SentenceDetector\n",
        "\n",
        "sent_detector = SentenceDetector()\\\n",
        "    .setInputCols(['document'])\\\n",
        "    .setOutputCol('sentences')"
      ],
      "execution_count": 68,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "SdJ5Nq-c97xF",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "b4d27239-f63d-4d80-b28c-1294a2dab17d"
      },
      "source": [
        "sentences = sent_detector.transform(docs)\n",
        "sentences.limit(5).toPandas()"
      ],
      "execution_count": 69,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>path</th>\n",
              "      <th>text</th>\n",
              "      <th>document</th>\n",
              "      <th>sentences</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!das-news.harva...</td>\n",
              "      <td>[(document, 0, 3758, Path: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 792, Path: cantaloupe.srv.cs.cm...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Newsgroups: sci.space\\nPath: cantaloupe.srv.cs...</td>\n",
              "      <td>[(document, 0, 1579, Newsgroups: sci.space\\nPa...</td>\n",
              "      <td>[(document, 0, 314, Newsgroups: sci.space\\nPat...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!magnesium.club...</td>\n",
              "      <td>[(document, 0, 4409, Path: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 810, Path: cantaloupe.srv.cs.cm...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Xref: cantaloupe.srv.cs.cmu.edu sci.space:6094...</td>\n",
              "      <td>[(document, 0, 2248, Xref: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 823, Xref: cantaloupe.srv.cs.cm...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!rochester!udel...</td>\n",
              "      <td>[(document, 0, 851, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(document, 0, 204, Path: cantaloupe.srv.cs.cm...</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "                                                path  ...                                          sentences\n",
              "0  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(document, 0, 792, Path: cantaloupe.srv.cs.cm...\n",
              "1  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(document, 0, 314, Newsgroups: sci.space\\nPat...\n",
              "2  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(document, 0, 810, Path: cantaloupe.srv.cs.cm...\n",
              "3  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(document, 0, 823, Xref: cantaloupe.srv.cs.cm...\n",
              "4  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(document, 0, 204, Path: cantaloupe.srv.cs.cm...\n",
              "\n",
              "[5 rows x 4 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 69
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "29gzaJ2Z995D",
        "colab_type": "text"
      },
      "source": [
        "#### Tokenizer\n",
        "\n",
        "A Tokenizer is a fundamental Annotator. Almost all text-based data processing begins with some form of tokenization. Most classical NLP algorithms expect tokens as the basic input. Many deep learning algorithms are being developed that take characters as basic input. Most NLP applications still use tokenization. The Spark NLP Tokenizer is a little more sophisticated than just a regular expression-based tokenizer. It has a number of parameters. The following are some of the basic ones (see table0316 for the results):\n",
        "\n",
        "* inputCols  \n",
        "A list of columns to tokenize.\n",
        "* outputCol  \n",
        "The name of the new token column.\n",
        "* targetPattern  \n",
        "Basic regex rule to identify a candidate for tokenization. Defaults to \\S+ which means anything not a space (optional).\n",
        "* prefixPattern  \n",
        "Regular expression (regex) to identify subtokens that come in the beginning of the token. Regex has to start with \\A and must contain groups (). Each group will become a separate token within the prefix. Defaults to nonletter characters—for example, quotes or parentheses (optional).\n",
        "* suffixPattern  \n",
        "Regex to identify subtokens that are in the end of the token. Regex has to end with \\z and must contain groups (). Each group will become a separate token within the prefix. Defaults to nonletter characters—for example, quotes or parentheses (optional)."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "QVupoF4n98wt",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from sparknlp.annotator import Tokenizer\n",
        "\n",
        "tokenizer = Tokenizer()\\\n",
        "    .setInputCols(['sentences'])\\\n",
        "    .setOutputCol('tokens')\\\n",
        "    .fit(sentences)"
      ],
      "execution_count": 70,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "gU3iypI_-JwE",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "078fed66-fb77-449e-f0f9-176e9784d36f"
      },
      "source": [
        "tokens = tokenizer.transform(sentences)\n",
        "tokens.limit(5).toPandas()"
      ],
      "execution_count": 71,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>path</th>\n",
              "      <th>text</th>\n",
              "      <th>document</th>\n",
              "      <th>sentences</th>\n",
              "      <th>tokens</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!das-news.harva...</td>\n",
              "      <td>[(document, 0, 3758, Path: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 792, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Newsgroups: sci.space\\nPath: cantaloupe.srv.cs...</td>\n",
              "      <td>[(document, 0, 1579, Newsgroups: sci.space\\nPa...</td>\n",
              "      <td>[(document, 0, 314, Newsgroups: sci.space\\nPat...</td>\n",
              "      <td>[(token, 0, 9, Newsgroups, {'sentence': '0'}, ...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!magnesium.club...</td>\n",
              "      <td>[(document, 0, 4409, Path: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 810, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Xref: cantaloupe.srv.cs.cmu.edu sci.space:6094...</td>\n",
              "      <td>[(document, 0, 2248, Xref: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 823, Xref: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Xref, {'sentence': '0'}, []), (...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!rochester!udel...</td>\n",
              "      <td>[(document, 0, 851, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(document, 0, 204, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "                                                path  ...                                             tokens\n",
              "0  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(token, 0, 3, Path, {'sentence': '0'}, []), (...\n",
              "1  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(token, 0, 9, Newsgroups, {'sentence': '0'}, ...\n",
              "2  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(token, 0, 3, Path, {'sentence': '0'}, []), (...\n",
              "3  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(token, 0, 3, Xref, {'sentence': '0'}, []), (...\n",
              "4  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(token, 0, 3, Path, {'sentence': '0'}, []), (...\n",
              "\n",
              "[5 rows x 5 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 71
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Zbowbqm6-Lwk",
        "colab_type": "text"
      },
      "source": [
        "There are some `Annotators` that require additional resources. Some require reference data, like the following example, the lemmatizer."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "khm7EB4C-QM-",
        "colab_type": "text"
      },
      "source": [
        "#### Lemmatizer\n",
        "\n",
        "The lemmatizer finds the lemmas for the tokens. Lemmas are the entry words in dictionaries. For example, \"cats\" lemmatizes to \"cat,\" and \"oxen\" lemmatizes to \"ox.\" Loading the lemmatizer requires a dictionary and the following three parameters:\n",
        "\n",
        "* inputCols  \n",
        "A list of columns to tokenize\n",
        "* outputCol  \n",
        "The name of the new token column\n",
        "* dictionary  \n",
        "The resource to be loaded as the lemma dictionary"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "k0eNjBa0-Kxk",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from sparknlp.annotator import Lemmatizer\n",
        "\n",
        "lemmatizer = Lemmatizer() \\\n",
        "  .setInputCols([\"tokens\"]) \\\n",
        "  .setOutputCol(\"lemma\") \\\n",
        "  .setDictionary('en_lemmas.txt', '\\t', ',')\\\n",
        "  .fit(tokens)"
      ],
      "execution_count": 75,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "jH-klgbV-Wpz",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 281
        },
        "outputId": "c56166aa-846f-4c47-b204-5b8b451d92f2"
      },
      "source": [
        "lemmas = lemmatizer.transform(tokens)\n",
        "lemmas.limit(5).toPandas()"
      ],
      "execution_count": 76,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>path</th>\n",
              "      <th>text</th>\n",
              "      <th>document</th>\n",
              "      <th>sentences</th>\n",
              "      <th>tokens</th>\n",
              "      <th>lemma</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!das-news.harva...</td>\n",
              "      <td>[(document, 0, 3758, Path: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 792, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Newsgroups: sci.space\\nPath: cantaloupe.srv.cs...</td>\n",
              "      <td>[(document, 0, 1579, Newsgroups: sci.space\\nPa...</td>\n",
              "      <td>[(document, 0, 314, Newsgroups: sci.space\\nPat...</td>\n",
              "      <td>[(token, 0, 9, Newsgroups, {'sentence': '0'}, ...</td>\n",
              "      <td>[(token, 0, 9, Newsgroups, {'sentence': '0'}, ...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!magnesium.club...</td>\n",
              "      <td>[(document, 0, 4409, Path: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 810, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Xref: cantaloupe.srv.cs.cmu.edu sci.space:6094...</td>\n",
              "      <td>[(document, 0, 2248, Xref: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 823, Xref: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Xref, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(token, 0, 3, Xref, {'sentence': '0'}, []), (...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!rochester!udel...</td>\n",
              "      <td>[(document, 0, 851, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(document, 0, 204, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "                                                path  ...                                              lemma\n",
              "0  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(token, 0, 3, Path, {'sentence': '0'}, []), (...\n",
              "1  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(token, 0, 9, Newsgroups, {'sentence': '0'}, ...\n",
              "2  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(token, 0, 3, Path, {'sentence': '0'}, []), (...\n",
              "3  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(token, 0, 3, Xref, {'sentence': '0'}, []), (...\n",
              "4  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(token, 0, 3, Path, {'sentence': '0'}, []), (...\n",
              "\n",
              "[5 rows x 6 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 76
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-tfE16_Q_Bmy",
        "colab_type": "text"
      },
      "source": [
        "#### POS tagger\n",
        "\n",
        "There are also Annotators that require models as resources. For example, the POS tagger uses a perceptron model, so it is called PerceptronApproach. The PerceptronApproach has five parameters:\n",
        "\n",
        "* inputCols  \n",
        "A list of columns to tag\n",
        "* outputCol  \n",
        "The name of the new tag column\n",
        "* posCol  \n",
        "Column of Array of POS tags that match tokens\n",
        "* corpus  \n",
        "POS tags delimited corpus; needs \"delimiter\" in options\n",
        "* nIterations  \n",
        "Number of iterations in training, converges to better accuracy\n",
        "\n",
        "We will load a pretrained model here. "
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "G5Y3jCTf-_0n",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from sparknlp.annotator import PerceptronModel"
      ],
      "execution_count": 77,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "JUd68mQ1_N_y",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 70
        },
        "outputId": "be1abe00-3b09-4467-ab6a-11c6534e265d"
      },
      "source": [
        "pos_tagger = PerceptronModel.pretrained() \\\n",
        "  .setInputCols([\"tokens\", \"sentences\"]) \\\n",
        "  .setOutputCol(\"pos\")"
      ],
      "execution_count": 78,
      "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": "code",
      "metadata": {
        "id": "lCFlMYyD_PBL",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 281
        },
        "outputId": "a4410bba-a387-490d-bf20-5979c38e0142"
      },
      "source": [
        "postags = pos_tagger.transform(lemmas)\n",
        "postags.limit(5).toPandas()"
      ],
      "execution_count": 79,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>path</th>\n",
              "      <th>text</th>\n",
              "      <th>document</th>\n",
              "      <th>sentences</th>\n",
              "      <th>tokens</th>\n",
              "      <th>lemma</th>\n",
              "      <th>pos</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!das-news.harva...</td>\n",
              "      <td>[(document, 0, 3758, Path: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 792, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(pos, 0, 3, NNP, {'word': 'Path'}, []), (pos,...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Newsgroups: sci.space\\nPath: cantaloupe.srv.cs...</td>\n",
              "      <td>[(document, 0, 1579, Newsgroups: sci.space\\nPa...</td>\n",
              "      <td>[(document, 0, 314, Newsgroups: sci.space\\nPat...</td>\n",
              "      <td>[(token, 0, 9, Newsgroups, {'sentence': '0'}, ...</td>\n",
              "      <td>[(token, 0, 9, Newsgroups, {'sentence': '0'}, ...</td>\n",
              "      <td>[(pos, 0, 9, NNP, {'word': 'Newsgroups'}, []),...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!magnesium.club...</td>\n",
              "      <td>[(document, 0, 4409, Path: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 810, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(pos, 0, 3, NNP, {'word': 'Path'}, []), (pos,...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Xref: cantaloupe.srv.cs.cmu.edu sci.space:6094...</td>\n",
              "      <td>[(document, 0, 2248, Xref: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 823, Xref: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Xref, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(token, 0, 3, Xref, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(pos, 0, 3, NNP, {'word': 'Xref'}, []), (pos,...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!rochester!udel...</td>\n",
              "      <td>[(document, 0, 851, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(document, 0, 204, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(pos, 0, 3, NNP, {'word': 'Path'}, []), (pos,...</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "                                                path  ...                                                pos\n",
              "0  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(pos, 0, 3, NNP, {'word': 'Path'}, []), (pos,...\n",
              "1  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(pos, 0, 9, NNP, {'word': 'Newsgroups'}, []),...\n",
              "2  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(pos, 0, 3, NNP, {'word': 'Path'}, []), (pos,...\n",
              "3  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(pos, 0, 3, NNP, {'word': 'Xref'}, []), (pos,...\n",
              "4  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(pos, 0, 3, NNP, {'word': 'Path'}, []), (pos,...\n",
              "\n",
              "[5 rows x 7 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 79
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lAAqz_pO_TLX",
        "colab_type": "text"
      },
      "source": [
        "### Pretrained Pipelines\n",
        "\n",
        "We saw earlier how we can organize multiple MLlib stages into a Pipeline. Using Pipelines is especially useful in NLP tasks because there are often many stages between loading the raw text and extracting structured data.\n",
        "\n",
        "Spark NLP has pretrained pipelines that can be used to process text. This doesn't mean that you do not need to tune pipelines for application. But it is often convenient to begin experimenting with a prebuilt NLP pipeline and find what needs tuning.\n",
        "\n",
        "#### Explain document ML pipeline\n",
        "\n",
        "The BasicPipeline does sentence splitting, tokenization, lemmatization, stemming, and POS tagging. If you want to get a quick look at some text data, this is a great pipeline to use"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "H8Yt5ByW_QVr",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 70
        },
        "outputId": "c83932e5-f4e8-4bf8-bf78-1b0310bf51eb"
      },
      "source": [
        "from sparknlp.pretrained import PretrainedPipeline\n",
        "\n",
        "pipeline = PretrainedPipeline('explain_document_ml', lang='en')"
      ],
      "execution_count": 80,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "explain_document_ml download started this may take some time.\n",
            "Approx size to download 9.4 MB\n",
            "[OK!]\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "aKC7kAht_dib",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 364
        },
        "outputId": "0e7c2ee3-1a84-4d6c-854a-ccabdc1b360e"
      },
      "source": [
        "pipeline.transform(texts).limit(5).toPandas()"
      ],
      "execution_count": 81,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>path</th>\n",
              "      <th>text</th>\n",
              "      <th>document</th>\n",
              "      <th>sentence</th>\n",
              "      <th>token</th>\n",
              "      <th>spell</th>\n",
              "      <th>lemmas</th>\n",
              "      <th>stems</th>\n",
              "      <th>pos</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!das-news.harva...</td>\n",
              "      <td>[(document, 0, 3758, Path: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 792, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0', 'confid...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0', 'confid...</td>\n",
              "      <td>[(token, 0, 3, path, {'sentence': '0', 'confid...</td>\n",
              "      <td>[(pos, 0, 3, NNP, {'word': 'Path'}, []), (pos,...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Newsgroups: sci.space\\nPath: cantaloupe.srv.cs...</td>\n",
              "      <td>[(document, 0, 1579, Newsgroups: sci.space\\nPa...</td>\n",
              "      <td>[(document, 0, 314, Newsgroups: sci.space\\nPat...</td>\n",
              "      <td>[(token, 0, 9, Newsgroups, {'sentence': '0'}, ...</td>\n",
              "      <td>[(token, 0, 9, Newsgroups, {'sentence': '0', '...</td>\n",
              "      <td>[(token, 0, 9, Newsgroups, {'sentence': '0', '...</td>\n",
              "      <td>[(token, 0, 9, newsgroup, {'sentence': '0', 'c...</td>\n",
              "      <td>[(pos, 0, 9, NNP, {'word': 'Newsgroups'}, []),...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!magnesium.club...</td>\n",
              "      <td>[(document, 0, 4409, Path: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 810, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0', 'confid...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0', 'confid...</td>\n",
              "      <td>[(token, 0, 3, path, {'sentence': '0', 'confid...</td>\n",
              "      <td>[(pos, 0, 3, NNP, {'word': 'Path'}, []), (pos,...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Xref: cantaloupe.srv.cs.cmu.edu sci.space:6094...</td>\n",
              "      <td>[(document, 0, 2248, Xref: cantaloupe.srv.cs.c...</td>\n",
              "      <td>[(document, 0, 823, Xref: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Xref, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(token, 0, 3, pref, {'sentence': '0', 'confid...</td>\n",
              "      <td>[(token, 0, 3, tref, {'sentence': '0', 'confid...</td>\n",
              "      <td>[(token, 0, 3, xref, {'sentence': '0', 'confid...</td>\n",
              "      <td>[(pos, 0, 3, NN, {'word': 'xref'}, []), (pos, ...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!rochester!udel...</td>\n",
              "      <td>[(document, 0, 851, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(document, 0, 204, Path: cantaloupe.srv.cs.cm...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0'}, []), (...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0', 'confid...</td>\n",
              "      <td>[(token, 0, 3, Path, {'sentence': '0', 'confid...</td>\n",
              "      <td>[(token, 0, 3, path, {'sentence': '0', 'confid...</td>\n",
              "      <td>[(pos, 0, 3, NNP, {'word': 'Path'}, []), (pos,...</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "                                                path  ...                                                pos\n",
              "0  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(pos, 0, 3, NNP, {'word': 'Path'}, []), (pos,...\n",
              "1  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(pos, 0, 9, NNP, {'word': 'Newsgroups'}, []),...\n",
              "2  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(pos, 0, 3, NNP, {'word': 'Path'}, []), (pos,...\n",
              "3  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(pos, 0, 3, NN, {'word': 'xref'}, []), (pos, ...\n",
              "4  file:/content/data/mini_newsgroups/sci.space/6...  ...  [(pos, 0, 3, NNP, {'word': 'Path'}, []), (pos,...\n",
              "\n",
              "[5 rows x 9 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 81
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "o82tgVLv_ees",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "text = texts.first()['text']"
      ],
      "execution_count": 82,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "fCvfeIzy_g8D",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 372
        },
        "outputId": "6df32850-cf6a-4476-eb80-675926364268"
      },
      "source": [
        "annotations = pipeline.annotate(text)\n",
        "list(zip(\n",
        "    annotations['token'], \n",
        "    annotations['stems'], \n",
        "    annotations['lemmas']\n",
        "))[100:120]"
      ],
      "execution_count": 83,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[('be', 'be', 'be'),\n",
              " ('carefully', 'carefulli', 'carefully'),\n",
              " ('constrained', 'constrain', 'constrain'),\n",
              " ('.', '.', '.'),\n",
              " ('.', '.', '.'),\n",
              " ('...', '...', '...'),\n",
              " ('No', 'no', 'No'),\n",
              " ('EDO', 'edo', 'EDO'),\n",
              " ('pallets', 'pallet', 'pallet'),\n",
              " (',', ',', ','),\n",
              " ('no', 'no', 'no'),\n",
              " ('spare', 'spare', 'spare'),\n",
              " ('Suits', 'suit', 'Suits'),\n",
              " (',', ',', ','),\n",
              " ('|>no', '|>no', '|>no'),\n",
              " ('extra', 'extra', 'extra'),\n",
              " (\"MMU's\", \"mmu'\", \"MMU's\"),\n",
              " ('.', '.', '.'),\n",
              " ('|', '|', '|'),\n",
              " ('|Has', 'ha', 'Has')]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 83
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "vNJkHpxA_jp8",
        "colab_type": "text"
      },
      "source": [
        "There are many other pipelines, and there is additional information [available](https://nlp.johnsnowlabs.com/docs/en/pipelines).\n",
        "\n",
        "Now let's talk about how we will perform step 4, converting the annotations into native Spark SQL types using the Finisher."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "JIm6p2iw_qM_",
        "colab_type": "text"
      },
      "source": [
        "### Finisher\n",
        "\n",
        "The annotations are useful for composing NLP steps, but we generally want to take some specific information out to process. The `Finisher` handles most of these use cases. If you want to get a list of tokens (or stems, or what have you) to use in downstream MLlib stages, the `Finisher` can do this (see table0320). Let's look at the parameters:\n",
        "\n",
        "* inputCols  \n",
        "Name of input annotation cols\n",
        "* outputCols  \n",
        "Name of finisher output cols\n",
        "* valueSplitSymbol  \n",
        "Character separating annotations\n",
        "* annotationSplitSymbol  \n",
        "Character separating annotations\n",
        "* cleanAnnotations  \n",
        "Determines whether to remove annotation columns\n",
        "* includeMetadata  \n",
        "Annotation metadata format\n",
        "* outputAsArray  \n",
        "Finisher generates an Array with the results instead of string"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "x0OGNNL4_iKC",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "finisher = Finisher()\\\n",
        "    .setInputCols(['tokens', 'lemma'])\\\n",
        "    .setOutputCols(['tokens', 'lemmata'])\\\n",
        "    .setCleanAnnotations(True)\\\n",
        "    .setOutputAsArray(True)"
      ],
      "execution_count": 84,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "1tJTzbG5_2na",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "custom_pipeline = Pipeline(stages=[\n",
        "    document_assembler,\n",
        "    sent_detector,\n",
        "    tokenizer,\n",
        "    lemmatizer,\n",
        "    finisher\n",
        "]).fit(texts)"
      ],
      "execution_count": 85,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "jq23jmJi_3mq",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "491ff008-9fb0-40f8-a8e4-df463f5b6dc5"
      },
      "source": [
        "custom_pipeline.transform(texts).limit(5).toPandas()"
      ],
      "execution_count": 86,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>path</th>\n",
              "      <th>text</th>\n",
              "      <th>tokens</th>\n",
              "      <th>lemmata</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!das-news.harva...</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!das-news.h...</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!das-news.h...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Newsgroups: sci.space\\nPath: cantaloupe.srv.cs...</td>\n",
              "      <td>[Newsgroups, :, sci.space, Path, :, cantaloupe...</td>\n",
              "      <td>[Newsgroups, :, sci.space, Path, :, cantaloupe...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!magnesium.club...</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!magnesium....</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!magnesium....</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Xref: cantaloupe.srv.cs.cmu.edu sci.space:6094...</td>\n",
              "      <td>[Xref, :, cantaloupe.srv.cs.cmu.edu, sci.space...</td>\n",
              "      <td>[Xref, :, cantaloupe.srv.cs.cmu.edu, sci.space...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!rochester!udel...</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!rochester!...</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!rochester!...</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "                                                path  ...                                            lemmata\n",
              "0  file:/content/data/mini_newsgroups/sci.space/6...  ...  [Path, :, cantaloupe.srv.cs.cmu.edu!das-news.h...\n",
              "1  file:/content/data/mini_newsgroups/sci.space/6...  ...  [Newsgroups, :, sci.space, Path, :, cantaloupe...\n",
              "2  file:/content/data/mini_newsgroups/sci.space/6...  ...  [Path, :, cantaloupe.srv.cs.cmu.edu!magnesium....\n",
              "3  file:/content/data/mini_newsgroups/sci.space/6...  ...  [Xref, :, cantaloupe.srv.cs.cmu.edu, sci.space...\n",
              "4  file:/content/data/mini_newsgroups/sci.space/6...  ...  [Path, :, cantaloupe.srv.cs.cmu.edu!rochester!...\n",
              "\n",
              "[5 rows x 4 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 86
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Sp02kNkt_40K",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.feature import StopWordsRemover"
      ],
      "execution_count": 87,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "jRnUE19R_6Nq",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "stopwords = StopWordsRemover.loadDefaultStopWords('english')"
      ],
      "execution_count": 88,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "H8l61JlZ_7ZC",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "larger_pipeline = Pipeline(stages=[\n",
        "    custom_pipeline,\n",
        "    StopWordsRemover(\n",
        "        inputCol='lemmata', \n",
        "        outputCol='terms', \n",
        "        stopWords=stopwords)\n",
        "]).fit(texts)"
      ],
      "execution_count": 89,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "PhPy7oyi_80K",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 197
        },
        "outputId": "0757436a-4324-4c81-c1ae-af7f1125915f"
      },
      "source": [
        "larger_pipeline.transform(texts).limit(5).toPandas()"
      ],
      "execution_count": 90,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "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>path</th>\n",
              "      <th>text</th>\n",
              "      <th>tokens</th>\n",
              "      <th>lemmata</th>\n",
              "      <th>terms</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!das-news.harva...</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!das-news.h...</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!das-news.h...</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!das-news.h...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Newsgroups: sci.space\\nPath: cantaloupe.srv.cs...</td>\n",
              "      <td>[Newsgroups, :, sci.space, Path, :, cantaloupe...</td>\n",
              "      <td>[Newsgroups, :, sci.space, Path, :, cantaloupe...</td>\n",
              "      <td>[Newsgroups, :, sci.space, Path, :, cantaloupe...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!magnesium.club...</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!magnesium....</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!magnesium....</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!magnesium....</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Xref: cantaloupe.srv.cs.cmu.edu sci.space:6094...</td>\n",
              "      <td>[Xref, :, cantaloupe.srv.cs.cmu.edu, sci.space...</td>\n",
              "      <td>[Xref, :, cantaloupe.srv.cs.cmu.edu, sci.space...</td>\n",
              "      <td>[Xref, :, cantaloupe.srv.cs.cmu.edu, sci.space...</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>file:/content/data/mini_newsgroups/sci.space/6...</td>\n",
              "      <td>Path: cantaloupe.srv.cs.cmu.edu!rochester!udel...</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!rochester!...</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!rochester!...</td>\n",
              "      <td>[Path, :, cantaloupe.srv.cs.cmu.edu!rochester!...</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "                                                path  ...                                              terms\n",
              "0  file:/content/data/mini_newsgroups/sci.space/6...  ...  [Path, :, cantaloupe.srv.cs.cmu.edu!das-news.h...\n",
              "1  file:/content/data/mini_newsgroups/sci.space/6...  ...  [Newsgroups, :, sci.space, Path, :, cantaloupe...\n",
              "2  file:/content/data/mini_newsgroups/sci.space/6...  ...  [Path, :, cantaloupe.srv.cs.cmu.edu!magnesium....\n",
              "3  file:/content/data/mini_newsgroups/sci.space/6...  ...  [Xref, :, cantaloupe.srv.cs.cmu.edu, sci.space...\n",
              "4  file:/content/data/mini_newsgroups/sci.space/6...  ...  [Path, :, cantaloupe.srv.cs.cmu.edu!rochester!...\n",
              "\n",
              "[5 rows x 5 columns]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 90
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "fste5GTS__PC",
        "colab_type": "text"
      },
      "source": [
        "Now that we have reviewed Spark and Spark NLP, we are almost ready to start building an NLP application. There is an extra benefit to learning an annotation library—it helps you understand how to structure a pipeline for NLP. This knowledge will be applicable even if you are using other technologies.\n",
        "\n",
        "The only topic left for us to cover is deep learning, which we cover in the next chapter.  "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "vteG8sRHABoj",
        "colab_type": "text"
      },
      "source": [
        "## Exercises: Build a Topic Model\n",
        "   One of the easiest things you can do to begin exploring your data set is to create a topic model. To do this we need to transform the text into numerical vectors. We will go into this more in the next part of the book. For now, let's build a pipeline for processing text.\n",
        "\n",
        "First, we need to split the texts into sentences. Second, we need to tokenize. Next, we need to normalize our words with a lemmatizer and the normalizer. After this, we need to finish our pipeline and remove stop words. (So far, everything except the Normalizer has been demonstrated in this chapter.) After that, we will pass the information into our topic modeling pipeline.\n",
        "\n",
        "Check the [online documentation](https://nlp.johnsnowlabs.com/docs/en/annotators#normalizer) for help with the normalizer."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "iWJCZYJi_-Ky",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# document_assembler = ???\n",
        "# sent_detector = ???\n",
        "# tokenizer = ???\n",
        "# lemmatizer = ???\n",
        "# normalizer = ???\n",
        "# finisher = ???\n",
        "# sparknlp_pipeline = ???"
      ],
      "execution_count": 91,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "1TGvmB2wANDa",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# stopwords = ???\n",
        "# stopword_remover = ??? # use outputCol='terms'\n",
        "\n",
        "#text_processing_pipeline = ??? # first stage is sparknlp_pipeline"
      ],
      "execution_count": 92,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "dp3hWHBXAOwK",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# from pyspark.ml.feature import CountVectorizer, IDF\n",
        "# from pyspark.ml.clustering import LDA\n",
        "\n",
        "# tf = CountVectorizer(inputCol='terms', outputCol='tf')\n",
        "# idf = IDF(inputCol='tf', outputCol='tfidf')\n",
        "# lda = LDA(k=10, seed=123, featuresCol='tfidf')\n",
        "\n",
        "# pipeline = Pipeline(stages=[\n",
        "#     text_processing_pipeline,\n",
        "#     tf,\n",
        "#     idf,\n",
        "#     lda\n",
        "# ])"
      ],
      "execution_count": 93,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "_X7MIcdfAQlr",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# model = pipeline.fit(texts)"
      ],
      "execution_count": 94,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "UNAMioU2ARmL",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# tf_model = model.stages[-3]\n",
        "# lda_model = model.stages[-1]"
      ],
      "execution_count": 95,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "WoyIS8AQASx6",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# topics = lda_model.describeTopics().collect()\n",
        "# for k, topic in enumerate(topics):\n",
        "#     print('Topic', k)\n",
        "#     for ix, wt in zip(topic['termIndices'], topic['termWeights']):\n",
        "#         print(ix, tf_model.vocabulary[ix], wt)\n",
        "#     print('#' * 50)"
      ],
      "execution_count": 96,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7p1GEHUjAUmT",
        "colab_type": "text"
      },
      "source": [
        "Congratulations! You have built your first full Spark pipeline with Spark NLP."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "639galEhAXTn",
        "colab_type": "text"
      },
      "source": [
        "Resources\n",
        "\n",
        "* Spark API: this is a great resource to have handy. With each new release there can be new additions, so I rely on the API documentation to keep up to date.\n",
        "  * Scala\n",
        "  * Java\n",
        "  * Python\n",
        "  * R\n",
        "  * Spark SQL programming guide\n",
        "  * MLlib programming guide\n",
        "* Natural Language Processing with Python: this is the book written explaining NLP with NLTK. It is available for free online.\n",
        "* spaCy: this library is doing a lot of cool things and has a strong community.\n",
        "* Spark NLP: the website of the Spark NLP library. Here you can find documentation, tutorials, and videos about the library.\n",
        "* Foundations of Statistical Natural Language Processing by Christopher D. Manning and Hinrich Schütze (MIT Press)\n",
        "  * This is considered a classic for natural language processing. It is an older book, so you won't find information on deep learning. If you want to have a strong foundation in NLP this is a great book to have.\n",
        "* Spark: The Definitive Guide by Matei Zaharia and Bill Chambers (O'Reilly)\n",
        "  * This is a great resource to have when using Spark.  "
      ]
    }
  ]
}