Yelp_NLP_pipeline.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Yelp_NLP_pipeline.ipynb",
      "provenance": [],
      "collapsed_sections": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    }
  },
  "cells": [
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "bDsVxaZ4BZ5u",
        "outputId": "b48e7e9f-52b8-4a5b-d7d8-a094f6dbace8"
      },
      "source": [
        "import os\r\n",
        "# Find the latest version of spark 3.0  from http://www-us.apache.org/dist/spark/ and enter as the spark version\r\n",
        "# For example:\r\n",
        "# spark_version = 'spark-3.0.2'\r\n",
        "spark_version = 'spark-3.0.2'\r\n",
        "os.environ['SPARK_VERSION']=spark_version\r\n",
        "\r\n",
        "# Install Spark and Java\r\n",
        "!apt-get update\r\n",
        "!apt-get install openjdk-11-jdk-headless -qq > /dev/null\r\n",
        "!wget -q http://www-us.apache.org/dist/spark/$SPARK_VERSION/$SPARK_VERSION-bin-hadoop2.7.tgz\r\n",
        "!tar xf $SPARK_VERSION-bin-hadoop2.7.tgz\r\n",
        "!pip install -q findspark\r\n",
        "\r\n",
        "# Set Environment Variables\r\n",
        "import os\r\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-11-openjdk-amd64\"\r\n",
        "os.environ[\"SPARK_HOME\"] = f\"/content/{spark_version}-bin-hadoop2.7\"\r\n",
        "\r\n",
        "# Start a SparkSession\r\n",
        "import findspark\r\n",
        "findspark.init()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Get:1 http://security.ubuntu.com/ubuntu bionic-security InRelease [88.7 kB]\n",
            "Ign:2 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  InRelease\n",
            "Get:3 https://cloud.r-project.org/bin/linux/ubuntu bionic-cran40/ InRelease [3,626 B]\n",
            "Hit:4 http://archive.ubuntu.com/ubuntu bionic InRelease\n",
            "Get:5 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic InRelease [15.9 kB]\n",
            "Ign:6 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  InRelease\n",
            "Get:7 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  Release [697 B]\n",
            "Hit:8 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  Release\n",
            "Get:9 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  Release.gpg [836 B]\n",
            "Get:10 http://archive.ubuntu.com/ubuntu bionic-updates InRelease [88.7 kB]\n",
            "Get:11 http://security.ubuntu.com/ubuntu bionic-security/universe amd64 Packages [1,396 kB]\n",
            "Hit:12 http://ppa.launchpad.net/cran/libgit2/ubuntu bionic InRelease\n",
            "Get:13 http://security.ubuntu.com/ubuntu bionic-security/main amd64 Packages [1,964 kB]\n",
            "Get:14 http://archive.ubuntu.com/ubuntu bionic-backports InRelease [74.6 kB]\n",
            "Hit:15 http://ppa.launchpad.net/deadsnakes/ppa/ubuntu bionic InRelease\n",
            "Hit:16 http://ppa.launchpad.net/graphics-drivers/ppa/ubuntu bionic InRelease\n",
            "Ign:18 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  Packages\n",
            "Get:18 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  Packages [602 kB]\n",
            "Get:19 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic/main Sources [1,745 kB]\n",
            "Get:20 http://archive.ubuntu.com/ubuntu bionic-updates/main amd64 Packages [2,394 kB]\n",
            "Get:21 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic/main amd64 Packages [893 kB]\n",
            "Get:22 http://archive.ubuntu.com/ubuntu bionic-updates/universe amd64 Packages [2,163 kB]\n",
            "Fetched 11.4 MB in 3s (3,720 kB/s)\n",
            "Reading package lists... Done\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GHBFIKrFBavp"
      },
      "source": [
        "# Start Spark session\r\n",
        "from pyspark.sql import SparkSession\r\n",
        "spark = SparkSession.builder.appName(\"Yelp_NLP\").getOrCreate()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "HRHtkyTxCf8f",
        "outputId": "0b877549-bf08-4ced-d83c-802fc7df9e8a"
      },
      "source": [
        "# Read in data from S3 Buckets\r\n",
        "from pyspark import SparkFiles\r\n",
        "url =\"https://s3.amazonaws.com/dataviz-curriculum/day_2/yelp_reviews.csv\"\r\n",
        "spark.sparkContext.addFile(url)\r\n",
        "df = spark.read.csv(SparkFiles.get(\"yelp_reviews.csv\"), sep=\",\", header=True)\r\n",
        "\r\n",
        "# Show DataFrame\r\n",
        "df.show()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+--------+--------------------+\n",
            "|   class|                text|\n",
            "+--------+--------------------+\n",
            "|positive|Wow... Loved this...|\n",
            "|negative|  Crust is not good.|\n",
            "|negative|Not tasty and the...|\n",
            "|positive|Stopped by during...|\n",
            "|positive|The selection on ...|\n",
            "|negative|Now I am getting ...|\n",
            "|negative|Honeslty it didn'...|\n",
            "|negative|The potatoes were...|\n",
            "|positive|The fries were gr...|\n",
            "|positive|      A great touch.|\n",
            "|positive|Service was very ...|\n",
            "|negative|  Would not go back.|\n",
            "|negative|The cashier had n...|\n",
            "|positive|I tried the Cape ...|\n",
            "|negative|I was disgusted b...|\n",
            "|negative|I was shocked bec...|\n",
            "|positive| Highly recommended.|\n",
            "|negative|Waitress was a li...|\n",
            "|negative|This place is not...|\n",
            "|negative|did not like at all.|\n",
            "+--------+--------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "BQOnxy_MCjY-"
      },
      "source": [
        "# Import functions\r\n",
        "from pyspark.ml.feature import Tokenizer, StopWordsRemover, HashingTF, IDF, StringIndexer"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "PaBPM4kHCvKC",
        "outputId": "00c38507-bec5-4e4d-eb4b-e9b8834734a0"
      },
      "source": [
        "from pyspark.sql.functions import length\r\n",
        "# Create a length column to be used as a future feature\r\n",
        "data_df = df.withColumn('length', length(df['text']))\r\n",
        "data_df.show()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+--------+--------------------+------+\n",
            "|   class|                text|length|\n",
            "+--------+--------------------+------+\n",
            "|positive|Wow... Loved this...|    24|\n",
            "|negative|  Crust is not good.|    18|\n",
            "|negative|Not tasty and the...|    41|\n",
            "|positive|Stopped by during...|    87|\n",
            "|positive|The selection on ...|    59|\n",
            "|negative|Now I am getting ...|    46|\n",
            "|negative|Honeslty it didn'...|    37|\n",
            "|negative|The potatoes were...|   111|\n",
            "|positive|The fries were gr...|    25|\n",
            "|positive|      A great touch.|    14|\n",
            "|positive|Service was very ...|    24|\n",
            "|negative|  Would not go back.|    18|\n",
            "|negative|The cashier had n...|    99|\n",
            "|positive|I tried the Cape ...|    59|\n",
            "|negative|I was disgusted b...|    62|\n",
            "|negative|I was shocked bec...|    50|\n",
            "|positive| Highly recommended.|    19|\n",
            "|negative|Waitress was a li...|    38|\n",
            "|negative|This place is not...|    51|\n",
            "|negative|did not like at all.|    20|\n",
            "+--------+--------------------+------+\n",
            "only showing top 20 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "bQEba7huDDOD"
      },
      "source": [
        "# Create all the features to the data set\r\n",
        "pos_neg_to_num = StringIndexer(inputCol='class', outputCol='label')\r\n",
        "tokenizer = Tokenizer(inputCol=\"text\", outputCol=\"token_text\")\r\n",
        "stopremover = StopWordsRemover(inputCol=\"token_text\", outputCol=\"stop_tokens\")\r\n",
        "hashingTF = HashingTF(inputCol=\"stop_tokens\", outputCol=\"hash_token\")\r\n",
        "idf = IDF(inputCol=\"hash_token\", outputCol=\"idf_token\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "d_sAN_YkDy5Z"
      },
      "source": [
        "from pyspark.ml.feature import VectorAssembler\r\n",
        "from pyspark.ml.linalg import Vector\r\n",
        "\r\n",
        "# Create feature vectors\r\n",
        "clean_up = VectorAssembler(inputCols=['idf_token', 'length'], outputCol='features')"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "d_3VM6o0EMca"
      },
      "source": [
        "# Create and run a data processing Pipeline\r\n",
        "from pyspark.ml import Pipeline\r\n",
        "data_prep_pipeline = Pipeline(stages=[pos_neg_to_num, tokenizer, stopremover, hashingTF, idf, clean_up])\r\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "cyoUQkkdEhZp"
      },
      "source": [
        "# Fit and transform the pipeline\r\n",
        "cleaner = data_prep_pipeline.fit(data_df)\r\n",
        "cleaned = cleaner.transform(data_df)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "d-L-WwVQFIiD",
        "outputId": "e8c9adaa-280b-4309-ee2a-836c12d20700"
      },
      "source": [
        "# Show label and resulting features\r\n",
        "cleaned.select(['label', 'features']).show(truncate=False)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+-----+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
            "|label|features                                                                                                                                                                                                                                                                     |\n",
            "+-----+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
            "|1.0  |(262145,[177414,216221,239331,262144],[6.215607598755275,4.51085950651685,3.8642323415917974,24.0])                                                                                                                                                                          |\n",
            "|0.0  |(262145,[49815,237498,262144],[3.7732605633860707,5.810142490647111,18.0])                                                                                                                                                                                                   |\n",
            "|0.0  |(262145,[109649,155665,223782,262144],[6.215607598755275,5.52246041819533,4.962844630259907,41.0])                                                                                                                                                                           |\n",
            "|1.0  |(262145,[53101,68727,91192,143501,146390,201874,216221,243871,248687,253047,262144],[6.215607598755275,6.215607598755275,5.116995310087166,4.075541435259004,6.215607598755275,5.810142490647111,4.51085950651685,6.215607598755275,6.215607598755275,5.52246041819533,87.0])|\n",
            "|1.0  |(262145,[15370,77733,151571,261870,262144],[4.829313237635384,5.52246041819533,4.6061696863211745,2.9197707327509463,59.0])                                                                                                                                                  |\n",
            "|0.0  |(262145,[98142,131537,190256,230764,255299,262144],[5.29931686688112,6.215607598755275,4.269697449699962,4.829313237635384,5.810142490647111,46.0])                                                                                                                          |\n",
            "|0.0  |(262145,[59172,221770,228557,262144],[6.215607598755275,4.711530201979001,6.215607598755275,37.0])                                                                                                                                                                           |\n",
            "|0.0  |(262145,[63420,85530,89720,121517,140784,159927,198468,208258,231798,262144],[6.215607598755275,4.962844630259907,5.810142490647111,3.3252358408591105,4.13616605707544,5.116995310087166,6.215607598755275,3.147554663621658,6.215607598755275,111.0])                      |\n",
            "|1.0  |(262145,[53777,171611,261870,262144],[5.29931686688112,4.829313237635384,2.9197707327509463,25.0])                                                                                                                                                                           |\n",
            "|1.0  |(262145,[221827,261870,262144],[6.215607598755275,2.9197707327509463,14.0])                                                                                                                                                                                                  |\n",
            "|1.0  |(262145,[43756,227670,262144],[2.883403088580071,6.215607598755275,24.0])                                                                                                                                                                                                    |\n",
            "|0.0  |(262145,[127310,148675,262144],[3.96431580014878,3.195182712610913,18.0])                                                                                                                                                                                                    |\n",
            "|0.0  |(262145,[407,31536,109230,171222,176797,185123,191174,203802,262144],[5.810142490647111,4.4238481295272205,5.810142490647111,4.343805421853684,5.810142490647111,6.215607598755275,5.810142490647111,3.730700948967275,99.0])                                                |\n",
            "|1.0  |(262145,[18098,93838,111292,148426,204491,210058,262144],[6.215607598755275,4.829313237635384,6.215607598755275,6.215607598755275,6.215607598755275,6.215607598755275,59.0])                                                                                                 |\n",
            "|0.0  |(262145,[23071,129900,134125,134303,248803,262144],[3.96431580014878,6.215607598755275,4.829313237635384,5.810142490647111,6.215607598755275,62.0])                                                                                                                          |\n",
            "|0.0  |(262145,[129941,159775,168139,182011,187580,262144],[6.215607598755275,6.215607598755275,5.810142490647111,6.215607598755275,6.215607598755275,50.0])                                                                                                                        |\n",
            "|1.0  |(262145,[19633,215473,262144],[5.29931686688112,6.215607598755275,19.0])                                                                                                                                                                                                     |\n",
            "|0.0  |(262145,[27707,65069,147752,242022,262144],[4.962844630259907,4.075541435259004,4.711530201979001,4.51085950651685,38.0])                                                                                                                                                    |\n",
            "|0.0  |(262145,[20891,27308,51247,70998,138954,173339,262144],[4.711530201979001,6.215607598755275,4.51085950651685,2.5780214390288894,6.215607598755275,5.810142490647111,51.0])                                                                                                   |\n",
            "|0.0  |(262145,[8287,208258,262144],[5.810142490647111,3.147554663621658,20.0])                                                                                                                                                                                                     |\n",
            "+-----+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "pUp2Qw6DFRub"
      },
      "source": [
        "# Break the data down into training set and testing set\r\n",
        "training, testing = cleaned.randomSplit([0.7, 0.3], 21)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "LCStPcbQGAqe"
      },
      "source": [
        "from pyspark.ml.classification import NaiveBayes\r\n",
        "# Create a Naive Bayes model and fit training data\r\n",
        "nb = NaiveBayes()\r\n",
        "predictor = nb.fit(training)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "W6EfWW9CGMYG",
        "outputId": "22fa7212-c59c-43fc-8e5e-cc39e8cd377b"
      },
      "source": [
        "# Transform the model with the testing data\r\n",
        "test_results = predictor.transform(testing)\r\n",
        "test_results.show(5)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+--------+--------------------+------+-----+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+----------+\n",
            "|   class|                text|length|label|          token_text|         stop_tokens|          hash_token|           idf_token|            features|       rawPrediction|         probability|prediction|\n",
            "+--------+--------------------+------+-----+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+----------+\n",
            "|negative|\"The burger... I ...|    86|  0.0|[\"the, burger...,...|[\"the, burger...,...|(262144,[20298,21...|(262144,[20298,21...|(262145,[20298,21...|[-820.60780566975...|[0.99999999999995...|       0.0|\n",
            "|negative|              #NAME?|     6|  0.0|            [#name?]|            [#name?]|(262144,[197050],...|(262144,[197050],...|(262145,[197050,2...|[-73.489435340867...|[0.07515735596910...|       1.0|\n",
            "|negative|After I pulled up...|    83|  0.0|[after, i, pulled...|[pulled, car, wai...|(262144,[65645,71...|(262144,[65645,71...|(262145,[65645,71...|[-620.40646705112...|[1.0,1.9205984091...|       0.0|\n",
            "|negative|Also, I feel like...|    58|  0.0|[also,, i, feel, ...|[also,, feel, lik...|(262144,[61899,66...|(262144,[61899,66...|(262145,[61899,66...|[-528.59562125515...|[0.99999999994873...|       0.0|\n",
            "|negative|Anyway, I do not ...|    44|  0.0|[anyway,, i, do, ...|[anyway,, think, ...|(262144,[132270,1...|(262144,[132270,1...|(262145,[132270,1...|[-334.09599709326...|[0.99999999994185...|       0.0|\n",
            "+--------+--------------------+------+-----+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------------+----------+\n",
            "only showing top 5 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "gNCxZHmzG7W0",
        "outputId": "179efeab-299a-4540-d3db-85dd122cd7d5"
      },
      "source": [
        "from pyspark.ml.evaluation import BinaryClassificationEvaluator\r\n",
        "\r\n",
        "acc_eval = BinaryClassificationEvaluator(labelCol='label', rawPredictionCol='prediction')\r\n",
        "acc = acc_eval.evaluate(test_results)\r\n",
        "print(\"Accuracy of model at predicting reviews was: %f\" % acc)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Accuracy of model at predicting reviews was: 0.700298\n"
          ],
          "name": "stdout"
        }
      ]
    }
  ]
}