Amazon_Reviews_Classification_Regression.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Amazon_Reviews_Classification_Regression.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "toc_visible": true,
      "authorship_tag": "ABX9TyMbv3/aUyDbOzLZpxQmvXYg",
      "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/susiexia/BigData_Amazon_reviews_ETL_Cloud/blob/master/Amazon_Reviews_Classification_Regression.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "rZnZaGYYgHNe",
        "colab_type": "text"
      },
      "source": [
        "# Project PART 3\n",
        "Perform ETL in the cloud AND analyze data using Natural Language Processing (NLP) pipeline including Machine Learning.\n",
        "- (**part 1** in \"Amazon_Reriews_ETL_process.ipynb\" <https://colab.research.google.com/drive/1N0fTd5rpGznaC15aYb5M63PDzCBJ_0e4>)\n",
        "\n",
        "- (**part 2** in \"Amazon_Reviews_NLP_ML.ipynb\" <https://colab.research.google.com/drive/1kAFj2v4wxlFVrCksN4CiKrWHC4dQj8f9>)\n",
        "\n",
        "**This part use the columns of helpful_votes and total_votes in vine dataframe to predict star_rating by linear regression model.**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "dhm2_J0FVOof",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Install Java, Spark, Findspark and download a Postgresql driver\n",
        "!apt-get install openjdk-8-jdk-headless -qq #> /dev/null\n",
        "!wget -q http://www-us.apache.org/dist/spark/spark-2.4.5/spark-2.4.5-bin-hadoop2.7.tgz\n",
        "!tar xf spark-2.4.5-bin-hadoop2.7.tgz\n",
        "!pip install -q findspark\n",
        "\n",
        "\n",
        "# Set Environment Variables\n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "os.environ[\"SPARK_HOME\"] = \"/content/spark-2.4.5-bin-hadoop2.7\"\n",
        "\n",
        "# Start a SparkSession\n",
        "import findspark\n",
        "findspark.init()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "E6utZrVsVRV7",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Create a spark session, configured with Posetgres driver\n",
        "from pyspark.sql import SparkSession\n",
        "spark = SparkSession.builder.appName('Amazon_LR').getOrCreate()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "e9WYZsSeVeNQ",
        "colab_type": "code",
        "outputId": "886d7358-ca62-496e-c30e-657ba0f0d76b",
        "colab": {
          "resources": {
            "http://localhost:8080/nbextensions/google.colab/files.js": {
              "data": "Ly8gQ29weXJpZ2h0IDIwMTcgR29vZ2xlIExMQwovLwovLyBMaWNlbnNlZCB1bmRlciB0aGUgQXBhY2hlIExpY2Vuc2UsIFZlcnNpb24gMi4wICh0aGUgIkxpY2Vuc2UiKTsKLy8geW91IG1heSBub3QgdXNlIHRoaXMgZmlsZSBleGNlcHQgaW4gY29tcGxpYW5jZSB3aXRoIHRoZSBMaWNlbnNlLgovLyBZb3UgbWF5IG9idGFpbiBhIGNvcHkgb2YgdGhlIExpY2Vuc2UgYXQKLy8KLy8gICAgICBodHRwOi8vd3d3LmFwYWNoZS5vcmcvbGljZW5zZXMvTElDRU5TRS0yLjAKLy8KLy8gVW5sZXNzIHJlcXVpcmVkIGJ5IGFwcGxpY2FibGUgbGF3IG9yIGFncmVlZCB0byBpbiB3cml0aW5nLCBzb2Z0d2FyZQovLyBkaXN0cmlidXRlZCB1bmRlciB0aGUgTGljZW5zZSBpcyBkaXN0cmlidXRlZCBvbiBhbiAiQVMgSVMiIEJBU0lTLAovLyBXSVRIT1VUIFdBUlJBTlRJRVMgT1IgQ09ORElUSU9OUyBPRiBBTlkgS0lORCwgZWl0aGVyIGV4cHJlc3Mgb3IgaW1wbGllZC4KLy8gU2VlIHRoZSBMaWNlbnNlIGZvciB0aGUgc3BlY2lmaWMgbGFuZ3VhZ2UgZ292ZXJuaW5nIHBlcm1pc3Npb25zIGFuZAovLyBsaW1pdGF0aW9ucyB1bmRlciB0aGUgTGljZW5zZS4KCi8qKgogKiBAZmlsZW92ZXJ2aWV3IEhlbHBlcnMgZm9yIGdvb2dsZS5jb2xhYiBQeXRob24gbW9kdWxlLgogKi8KKGZ1bmN0aW9uKHNjb3BlKSB7CmZ1bmN0aW9uIHNwYW4odGV4dCwgc3R5bGVBdHRyaWJ1dGVzID0ge30pIHsKICBjb25zdCBlbGVtZW50ID0gZG9jdW1lbnQuY3JlYXRlRWxlbWVudCgnc3BhbicpOwogIGVsZW1lbnQudGV4dENvbnRlbnQgPSB0ZXh0OwogIGZvciAoY29uc3Qga2V5IG9mIE9iamVjdC5rZXlzKHN0eWxlQXR0cmlidXRlcykpIHsKICAgIGVsZW1lbnQuc3R5bGVba2V5XSA9IHN0eWxlQXR0cmlidXRlc1trZXldOwogIH0KICByZXR1cm4gZWxlbWVudDsKfQoKLy8gTWF4IG51bWJlciBvZiBieXRlcyB3aGljaCB3aWxsIGJlIHVwbG9hZGVkIGF0IGEgdGltZS4KY29uc3QgTUFYX1BBWUxPQURfU0laRSA9IDEwMCAqIDEwMjQ7Ci8vIE1heCBhbW91bnQgb2YgdGltZSB0byBibG9jayB3YWl0aW5nIGZvciB0aGUgdXNlci4KY29uc3QgRklMRV9DSEFOR0VfVElNRU9VVF9NUyA9IDMwICogMTAwMDsKCmZ1bmN0aW9uIF91cGxvYWRGaWxlcyhpbnB1dElkLCBvdXRwdXRJZCkgewogIGNvbnN0IHN0ZXBzID0gdXBsb2FkRmlsZXNTdGVwKGlucHV0SWQsIG91dHB1dElkKTsKICBjb25zdCBvdXRwdXRFbGVtZW50ID0gZG9jdW1lbnQuZ2V0RWxlbWVudEJ5SWQob3V0cHV0SWQpOwogIC8vIENhY2hlIHN0ZXBzIG9uIHRoZSBvdXRwdXRFbGVtZW50IHRvIG1ha2UgaXQgYXZhaWxhYmxlIGZvciB0aGUgbmV4dCBjYWxsCiAgLy8gdG8gdXBsb2FkRmlsZXNDb250aW51ZSBmcm9tIFB5dGhvbi4KICBvdXRwdXRFbGVtZW50LnN0ZXBzID0gc3RlcHM7CgogIHJldHVybiBfdXBsb2FkRmlsZXNDb250aW51ZShvdXRwdXRJZCk7Cn0KCi8vIFRoaXMgaXMgcm91Z2hseSBhbiBhc3luYyBnZW5lcmF0b3IgKG5vdCBzdXBwb3J0ZWQgaW4gdGhlIGJyb3dzZXIgeWV0KSwKLy8gd2hlcmUgdGhlcmUgYXJlIG11bHRpcGxlIGFzeW5jaHJvbm91cyBzdGVwcyBhbmQgdGhlIFB5dGhvbiBzaWRlIGlzIGdvaW5nCi8vIHRvIHBvbGwgZm9yIGNvbXBsZXRpb24gb2YgZWFjaCBzdGVwLgovLyBUaGlzIHVzZXMgYSBQcm9taXNlIHRvIGJsb2NrIHRoZSBweXRob24gc2lkZSBvbiBjb21wbGV0aW9uIG9mIGVhY2ggc3RlcCwKLy8gdGhlbiBwYXNzZXMgdGhlIHJlc3VsdCBvZiB0aGUgcHJldmlvdXMgc3RlcCBhcyB0aGUgaW5wdXQgdG8gdGhlIG5leHQgc3RlcC4KZnVuY3Rpb24gX3VwbG9hZEZpbGVzQ29udGludWUob3V0cHV0SWQpIHsKICBjb25zdCBvdXRwdXRFbGVtZW50ID0gZG9jdW1lbnQuZ2V0RWxlbWVudEJ5SWQob3V0cHV0SWQpOwogIGNvbnN0IHN0ZXBzID0gb3V0cHV0RWxlbWVudC5zdGVwczsKCiAgY29uc3QgbmV4dCA9IHN0ZXBzLm5leHQob3V0cHV0RWxlbWVudC5sYXN0UHJvbWlzZVZhbHVlKTsKICByZXR1cm4gUHJvbWlzZS5yZXNvbHZlKG5leHQudmFsdWUucHJvbWlzZSkudGhlbigodmFsdWUpID0+IHsKICAgIC8vIENhY2hlIHRoZSBsYXN0IHByb21pc2UgdmFsdWUgdG8gbWFrZSBpdCBhdmFpbGFibGUgdG8gdGhlIG5leHQKICAgIC8vIHN0ZXAgb2YgdGhlIGdlbmVyYXRvci4KICAgIG91dHB1dEVsZW1lbnQubGFzdFByb21pc2VWYWx1ZSA9IHZhbHVlOwogICAgcmV0dXJuIG5leHQudmFsdWUucmVzcG9uc2U7CiAgfSk7Cn0KCi8qKgogKiBHZW5lcmF0b3IgZnVuY3Rpb24gd2hpY2ggaXMgY2FsbGVkIGJldHdlZW4gZWFjaCBhc3luYyBzdGVwIG9mIHRoZSB1cGxvYWQKICogcHJvY2Vzcy4KICogQHBhcmFtIHtzdHJpbmd9IGlucHV0SWQgRWxlbWVudCBJRCBvZiB0aGUgaW5wdXQgZmlsZSBwaWNrZXIgZWxlbWVudC4KICogQHBhcmFtIHtzdHJpbmd9IG91dHB1dElkIEVsZW1lbnQgSUQgb2YgdGhlIG91dHB1dCBkaXNwbGF5LgogKiBAcmV0dXJuIHshSXRlcmFibGU8IU9iamVjdD59IEl0ZXJhYmxlIG9mIG5leHQgc3RlcHMuCiAqLwpmdW5jdGlvbiogdXBsb2FkRmlsZXNTdGVwKGlucHV0SWQsIG91dHB1dElkKSB7CiAgY29uc3QgaW5wdXRFbGVtZW50ID0gZG9jdW1lbnQuZ2V0RWxlbWVudEJ5SWQoaW5wdXRJZCk7CiAgaW5wdXRFbGVtZW50LmRpc2FibGVkID0gZmFsc2U7CgogIGNvbnN0IG91dHB1dEVsZW1lbnQgPSBkb2N1bWVudC5nZXRFbGVtZW50QnlJZChvdXRwdXRJZCk7CiAgb3V0cHV0RWxlbWVudC5pbm5lckhUTUwgPSAnJzsKCiAgY29uc3QgcGlja2VkUHJvbWlzZSA9IG5ldyBQcm9taXNlKChyZXNvbHZlKSA9PiB7CiAgICBpbnB1dEVsZW1lbnQuYWRkRXZlbnRMaXN0ZW5lcignY2hhbmdlJywgKGUpID0+IHsKICAgICAgcmVzb2x2ZShlLnRhcmdldC5maWxlcyk7CiAgICB9KTsKICB9KTsKCiAgY29uc3QgY2FuY2VsID0gZG9jdW1lbnQuY3JlYXRlRWxlbWVudCgnYnV0dG9uJyk7CiAgaW5wdXRFbGVtZW50LnBhcmVudEVsZW1lbnQuYXBwZW5kQ2hpbGQoY2FuY2VsKTsKICBjYW5jZWwudGV4dENvbnRlbnQgPSAnQ2FuY2VsIHVwbG9hZCc7CiAgY29uc3QgY2FuY2VsUHJvbWlzZSA9IG5ldyBQcm9taXNlKChyZXNvbHZlKSA9PiB7CiAgICBjYW5jZWwub25jbGljayA9ICgpID0+IHsKICAgICAgcmVzb2x2ZShudWxsKTsKICAgIH07CiAgfSk7CgogIC8vIENhbmNlbCB1cGxvYWQgaWYgdXNlciBoYXNuJ3QgcGlja2VkIGFueXRoaW5nIGluIHRpbWVvdXQuCiAgY29uc3QgdGltZW91dFByb21pc2UgPSBuZXcgUHJvbWlzZSgocmVzb2x2ZSkgPT4gewogICAgc2V0VGltZW91dCgoKSA9PiB7CiAgICAgIHJlc29sdmUobnVsbCk7CiAgICB9LCBGSUxFX0NIQU5HRV9USU1FT1VUX01TKTsKICB9KTsKCiAgLy8gV2FpdCBmb3IgdGhlIHVzZXIgdG8gcGljayB0aGUgZmlsZXMuCiAgY29uc3QgZmlsZXMgPSB5aWVsZCB7CiAgICBwcm9taXNlOiBQcm9taXNlLnJhY2UoW3BpY2tlZFByb21pc2UsIHRpbWVvdXRQcm9taXNlLCBjYW5jZWxQcm9taXNlXSksCiAgICByZXNwb25zZTogewogICAgICBhY3Rpb246ICdzdGFydGluZycsCiAgICB9CiAgfTsKCiAgaWYgKCFmaWxlcykgewogICAgcmV0dXJuIHsKICAgICAgcmVzcG9uc2U6IHsKICAgICAgICBhY3Rpb246ICdjb21wbGV0ZScsCiAgICAgIH0KICAgIH07CiAgfQoKICBjYW5jZWwucmVtb3ZlKCk7CgogIC8vIERpc2FibGUgdGhlIGlucHV0IGVsZW1lbnQgc2luY2UgZnVydGhlciBwaWNrcyBhcmUgbm90IGFsbG93ZWQuCiAgaW5wdXRFbGVtZW50LmRpc2FibGVkID0gdHJ1ZTsKCiAgZm9yIChjb25zdCBmaWxlIG9mIGZpbGVzKSB7CiAgICBjb25zdCBsaSA9IGRvY3VtZW50LmNyZWF0ZUVsZW1lbnQoJ2xpJyk7CiAgICBsaS5hcHBlbmQoc3BhbihmaWxlLm5hbWUsIHtmb250V2VpZ2h0OiAnYm9sZCd9KSk7CiAgICBsaS5hcHBlbmQoc3BhbigKICAgICAgICBgKCR7ZmlsZS50eXBlIHx8ICduL2EnfSkgLSAke2ZpbGUuc2l6ZX0gYnl0ZXMsIGAgKwogICAgICAgIGBsYXN0IG1vZGlmaWVkOiAkewogICAgICAgICAgICBmaWxlLmxhc3RNb2RpZmllZERhdGUgPyBmaWxlLmxhc3RNb2RpZmllZERhdGUudG9Mb2NhbGVEYXRlU3RyaW5nKCkgOgogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAnbi9hJ30gLSBgKSk7CiAgICBjb25zdCBwZXJjZW50ID0gc3BhbignMCUgZG9uZScpOwogICAgbGkuYXBwZW5kQ2hpbGQocGVyY2VudCk7CgogICAgb3V0cHV0RWxlbWVudC5hcHBlbmRDaGlsZChsaSk7CgogICAgY29uc3QgZmlsZURhdGFQcm9taXNlID0gbmV3IFByb21pc2UoKHJlc29sdmUpID0+IHsKICAgICAgY29uc3QgcmVhZGVyID0gbmV3IEZpbGVSZWFkZXIoKTsKICAgICAgcmVhZGVyLm9ubG9hZCA9IChlKSA9PiB7CiAgICAgICAgcmVzb2x2ZShlLnRhcmdldC5yZXN1bHQpOwogICAgICB9OwogICAgICByZWFkZXIucmVhZEFzQXJyYXlCdWZmZXIoZmlsZSk7CiAgICB9KTsKICAgIC8vIFdhaXQgZm9yIHRoZSBkYXRhIHRvIGJlIHJlYWR5LgogICAgbGV0IGZpbGVEYXRhID0geWllbGQgewogICAgICBwcm9taXNlOiBmaWxlRGF0YVByb21pc2UsCiAgICAgIHJlc3BvbnNlOiB7CiAgICAgICAgYWN0aW9uOiAnY29udGludWUnLAogICAgICB9CiAgICB9OwoKICAgIC8vIFVzZSBhIGNodW5rZWQgc2VuZGluZyB0byBhdm9pZCBtZXNzYWdlIHNpemUgbGltaXRzLiBTZWUgYi82MjExNTY2MC4KICAgIGxldCBwb3NpdGlvbiA9IDA7CiAgICB3aGlsZSAocG9zaXRpb24gPCBmaWxlRGF0YS5ieXRlTGVuZ3RoKSB7CiAgICAgIGNvbnN0IGxlbmd0aCA9IE1hdGgubWluKGZpbGVEYXRhLmJ5dGVMZW5ndGggLSBwb3NpdGlvbiwgTUFYX1BBWUxPQURfU0laRSk7CiAgICAgIGNvbnN0IGNodW5rID0gbmV3IFVpbnQ4QXJyYXkoZmlsZURhdGEsIHBvc2l0aW9uLCBsZW5ndGgpOwogICAgICBwb3NpdGlvbiArPSBsZW5ndGg7CgogICAgICBjb25zdCBiYXNlNjQgPSBidG9hKFN0cmluZy5mcm9tQ2hhckNvZGUuYXBwbHkobnVsbCwgY2h1bmspKTsKICAgICAgeWllbGQgewogICAgICAgIHJlc3BvbnNlOiB7CiAgICAgICAgICBhY3Rpb246ICdhcHBlbmQnLAogICAgICAgICAgZmlsZTogZmlsZS5uYW1lLAogICAgICAgICAgZGF0YTogYmFzZTY0LAogICAgICAgIH0sCiAgICAgIH07CiAgICAgIHBlcmNlbnQudGV4dENvbnRlbnQgPQogICAgICAgICAgYCR7TWF0aC5yb3VuZCgocG9zaXRpb24gLyBmaWxlRGF0YS5ieXRlTGVuZ3RoKSAqIDEwMCl9JSBkb25lYDsKICAgIH0KICB9CgogIC8vIEFsbCBkb25lLgogIHlpZWxkIHsKICAgIHJlc3BvbnNlOiB7CiAgICAgIGFjdGlvbjogJ2NvbXBsZXRlJywKICAgIH0KICB9Owp9CgpzY29wZS5nb29nbGUgPSBzY29wZS5nb29nbGUgfHwge307CnNjb3BlLmdvb2dsZS5jb2xhYiA9IHNjb3BlLmdvb2dsZS5jb2xhYiB8fCB7fTsKc2NvcGUuZ29vZ2xlLmNvbGFiLl9maWxlcyA9IHsKICBfdXBsb2FkRmlsZXMsCiAgX3VwbG9hZEZpbGVzQ29udGludWUsCn07Cn0pKHNlbGYpOwo=",
              "ok": true,
              "headers": [
                [
                  "content-type",
                  "application/javascript"
                ]
              ],
              "status": 200,
              "status_text": ""
            }
          },
          "base_uri": "https://localhost:8080/",
          "height": 40
        }
      },
      "source": [
        "from google.colab import files\n",
        "uploaded = files.upload()"
      ],
      "execution_count": 8,
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "text/html": [
              "\n",
              "     <input type=\"file\" id=\"files-a3da319e-bb3a-4b87-b1b2-4eac39b00b1b\" name=\"files[]\" multiple disabled />\n",
              "     <output id=\"result-a3da319e-bb3a-4b87-b1b2-4eac39b00b1b\">\n",
              "      Upload widget is only available when the cell has been executed in the\n",
              "      current browser session. Please rerun this cell to enable.\n",
              "      </output>\n",
              "      <script src=\"/nbextensions/google.colab/files.js\"></script> "
            ],
            "text/plain": [
              "<IPython.core.display.HTML object>"
            ]
          },
          "metadata": {
            "tags": []
          }
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "uvVEXuF2aJKm",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.sql.types import StructField, StructType, IntegerType, StringType\n",
        "schema = [StructField('review_id',StringType(),True), StructField('star_rating', IntegerType(), True), \n",
        "          StructField('helpful_votes', IntegerType(), True), StructField('total_votes', IntegerType(), True),\n",
        "          StructField('vine',StringType(),True)]\n",
        "final = StructType(fields=schema)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "F3bmkseoVpGL",
        "colab_type": "code",
        "outputId": "9655a6ce-c84a-43f0-fc04-abe83264efa0",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 476
        }
      },
      "source": [
        "vine_reviews_df = spark.read.csv('vine.csv', sep=',',header=True, schema=final)\n",
        "vine_reviews_df.show()\n",
        "vine_reviews_df.printSchema\n"
      ],
      "execution_count": 9,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+--------------+-----------+-------------+-----------+----+\n",
            "|     review_id|star_rating|helpful_votes|total_votes|vine|\n",
            "+--------------+-----------+-------------+-----------+----+\n",
            "| RWOE9SUY9N8J7|          5|            0|          0|   N|\n",
            "|R3QRJQDHI4N1NW|          1|            4|          4|   N|\n",
            "| RS02YZ0GSKWJJ|          2|            1|          1|   N|\n",
            "|R2EIZK8D31VYZO|          3|            0|          1|   N|\n",
            "|R23E01JXNIQRXI|          3|            0|          1|   N|\n",
            "|R1ZGQ80LY59OEK|          5|            3|          3|   N|\n",
            "|R1JICQZYMO0IM0|          5|            0|          0|   N|\n",
            "|R2COOCGQX2QXNB|          5|            0|          1|   N|\n",
            "|R28KX9E8RB627P|          1|            2|          3|   N|\n",
            "|R1IIDFQLB2TVLQ|          4|            3|          3|   N|\n",
            "|R3H6FT9FPDZI6C|          5|            0|          1|   N|\n",
            "| RMXCHZNGBWHRO|          5|            1|          1|   N|\n",
            "| R6MEWP6M0LG5D|          5|            0|          0|   N|\n",
            "|R2KDB9Y7EBBYQB|          5|            2|          2|   N|\n",
            "|R16R7CS6P2JH8A|          5|            0|          0|   N|\n",
            "|R2AOCXQ7HIZ0PT|          4|            1|          1|   N|\n",
            "|R3CSM8BOLP6SQW|          5|            0|          0|   N|\n",
            "|R394QJBA00VNWJ|          5|           18|         22|   N|\n",
            "|R1EK7MYNKYM6Q3|          3|            9|         10|   N|\n",
            "|R2FKB55ITMJFUB|          5|            2|          3|   N|\n",
            "+--------------+-----------+-------------+-----------+----+\n",
            "only showing top 20 rows\n",
            "\n"
          ],
          "name": "stdout"
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "<bound method DataFrame.printSchema of DataFrame[review_id: string, star_rating: int, helpful_votes: int, total_votes: int, vine: string]>"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 9
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "mCIb34JJicMG",
        "colab_type": "code",
        "outputId": "b4ec92ff-d098-45ff-b02e-e0964baa0a73",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 204
        }
      },
      "source": [
        "# add a column for helpful votes percentage\n",
        "from pyspark.sql.types import IntegerType\n",
        "from pyspark.sql import Column\n",
        "\n",
        "vine_reviews_df = vine_reviews_df.withColumn('helpful_votes_percentage', \n",
        "                                             vine_reviews_df['helpful_votes']/vine_reviews_df['total_votes'])\n",
        "\n",
        "\n",
        "vine_reviews_df = vine_reviews_df.dropna()\n",
        "vine_reviews_df.show(5)"
      ],
      "execution_count": 10,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+--------------+-----------+-------------+-----------+----+------------------------+\n",
            "|     review_id|star_rating|helpful_votes|total_votes|vine|helpful_votes_percentage|\n",
            "+--------------+-----------+-------------+-----------+----+------------------------+\n",
            "|R3QRJQDHI4N1NW|          1|            4|          4|   N|                     1.0|\n",
            "| RS02YZ0GSKWJJ|          2|            1|          1|   N|                     1.0|\n",
            "|R2EIZK8D31VYZO|          3|            0|          1|   N|                     0.0|\n",
            "|R23E01JXNIQRXI|          3|            0|          1|   N|                     0.0|\n",
            "|R1ZGQ80LY59OEK|          5|            3|          3|   N|                     1.0|\n",
            "+--------------+-----------+-------------+-----------+----+------------------------+\n",
            "only showing top 5 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "iS7dchByqYV9",
        "colab_type": "code",
        "outputId": "02a57546-1cc1-43bc-e6cb-a0821451a745",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 187
        }
      },
      "source": [
        "# statistical summary \n",
        "vine_reviews_df.describe().show()"
      ],
      "execution_count": 11,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+-------+--------------+------------------+-----------------+------------------+------+------------------------+\n",
            "|summary|     review_id|       star_rating|    helpful_votes|       total_votes|  vine|helpful_votes_percentage|\n",
            "+-------+--------------+------------------+-----------------+------------------+------+------------------------+\n",
            "|  count|        353555|            353555|           353555|            353555|353555|                  353555|\n",
            "|   mean|          null|3.9592962905347115|7.058344529139738| 8.554001499059552|  null|      0.7606408731303688|\n",
            "| stddev|          null|1.4528450865994198|46.76890783473558|48.930514112084055|  null|      0.3220946950293199|\n",
            "|    min|R1001H341IC645|                 1|                0|                 1|     N|                     0.0|\n",
            "|    max| RZZZPYCL9LDIT|                 5|            13362|             13636|     Y|                     1.0|\n",
            "+-------+--------------+------------------+-----------------+------------------+------+------------------------+\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "s3TjeTAeZOZV",
        "colab_type": "text"
      },
      "source": [
        "# Naive Bayes\n",
        "\n",
        "Results: There will be 0.97 accuracy of Naive Bayes to determine if a review will be vine reviews based on helpful_votes, total_votes and star_rating."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ViQ8NhnMZVB7",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 204
        },
        "outputId": "c026bfe1-c974-446d-c95d-bc77edbd91a6"
      },
      "source": [
        "# convert vine column to dichotomous types\n",
        "from pyspark.ml.feature import StringIndexer\n",
        "from pyspark.ml.feature import VectorAssembler\n",
        "from pyspark.ml.classification import NaiveBayes\n",
        "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
        "from pyspark.ml import Pipeline\n",
        "\n",
        "strIndexed = StringIndexer(inputCol='vine', outputCol='label')\n",
        "\n",
        "assembler = VectorAssembler(inputCols=['helpful_votes', 'total_votes','star_rating'], outputCol='Attributes')\n",
        "\n",
        "data_prep_pipeline = Pipeline(stages= [strIndexed, assembler])\n",
        "\n",
        "pipeline = data_prep_pipeline.fit(vine_reviews_df)\n",
        "lr_df = pipeline.transform(vine_reviews_df)\n",
        "clean_df = lr_df.select(\"Attributes\",\"label\")\n",
        "clean_df.show(5)"
      ],
      "execution_count": 13,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+-------------+-----+\n",
            "|   Attributes|label|\n",
            "+-------------+-----+\n",
            "|[4.0,4.0,1.0]|  0.0|\n",
            "|[1.0,1.0,2.0]|  0.0|\n",
            "|[0.0,1.0,3.0]|  0.0|\n",
            "|[0.0,1.0,3.0]|  0.0|\n",
            "|[3.0,3.0,5.0]|  0.0|\n",
            "+-------------+-----+\n",
            "only showing top 5 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "YwqPpHjfZwIj",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 459
        },
        "outputId": "dfdcbb68-965d-4777-8257-bfdb6c3cd4cd"
      },
      "source": [
        "# break whole data down into a training set and a testing set\n",
        "training, testing = clean_df.randomSplit([0.7,0.3])\n",
        "\n",
        "#create a Naive Bayes Model \n",
        "nb = NaiveBayes(featuresCol='Attributes', labelCol='label') \n",
        "predictor = nb.fit(training)     # fit training df to nb model, predictor is NaiveBayes object\n",
        "\n",
        "# transform the model with teasting data\n",
        "test_results = predictor.transform(testing)\n",
        "test_results.select('Attributes', 'rawPrediction','probability','prediction').show(truncate= False)\n"
      ],
      "execution_count": 16,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+-------------+---------------------------------------+-----------------------------------------+----------+\n",
            "|Attributes   |rawPrediction                          |probability                              |prediction|\n",
            "+-------------+---------------------------------------+-----------------------------------------+----------+\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "|[0.0,1.0,1.0]|[-2.428943926370085,-7.120069394144616]|[0.9909070871220875,0.009092912877912376]|0.0       |\n",
            "+-------------+---------------------------------------+-----------------------------------------+----------+\n",
            "only showing top 20 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "6QCqas08aful",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 34
        },
        "outputId": "16e6b839-6c41-42f5-c593-18abb7ea900a"
      },
      "source": [
        "# use the Class Evaluator for a cleaner description\n",
        "acc_eval = MulticlassClassificationEvaluator()\n",
        "\n",
        "acc = acc_eval.evaluate(test_results)   \n",
        "\n",
        "print(\"Accuracy of model at predicting vine reviews was : %f \"% acc)"
      ],
      "execution_count": 18,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Accuracy of model at predicting vine reviews was : 0.976541 \n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "CKgh82ZerQHf",
        "colab_type": "text"
      },
      "source": [
        "# Linear Regression"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "OCkIES3QYdYs",
        "colab_type": "code",
        "outputId": "e3ba9231-93b4-433f-9558-6502b8dd71bc",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 221
        }
      },
      "source": [
        "from pyspark.ml.feature import VectorAssembler\n",
        "from pyspark.ml.regression import LinearRegression\n",
        "\n",
        "assembler = VectorAssembler(inputCols=['helpful_votes', 'total_votes'], outputCol='Attributes')\n",
        "\n",
        "assembed_df = assembler.transform(vine_reviews_df)\n",
        "\n",
        "finalized_data = assembed_df.select(\"Attributes\",\"star_rating\")\n",
        "finalized_data.show(5)\n",
        "finalized_data.count()\n"
      ],
      "execution_count": 0,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+----------+-----------+\n",
            "|Attributes|star_rating|\n",
            "+----------+-----------+\n",
            "| [4.0,4.0]|          1|\n",
            "| [1.0,1.0]|          2|\n",
            "| [0.0,1.0]|          3|\n",
            "| [0.0,1.0]|          3|\n",
            "| [3.0,3.0]|          5|\n",
            "+----------+-----------+\n",
            "only showing top 5 rows\n",
            "\n"
          ],
          "name": "stdout"
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "2219424"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 36
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "EGEGDmozd893",
        "colab_type": "code",
        "outputId": "128d61e5-0547-4d91-d169-ee9971a92d5c",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 204
        }
      },
      "source": [
        "#Split training and testing data\n",
        "training,testing = finalized_data.randomSplit([0.7,0.3])\n",
        "\n",
        "regressor = LinearRegression(featuresCol = 'Attributes', labelCol = 'star_rating')\n",
        "#Learn to fit the model from training set\n",
        "regressor = regressor.fit(training)\n",
        "#To predict the prices on testing set\n",
        "predicting = regressor.evaluate(testing)\n",
        "#Predict the model\n",
        "predicting.predictions.show(5)"
      ],
      "execution_count": 0,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+----------+-----------+-----------------+\n",
            "|Attributes|star_rating|       prediction|\n",
            "+----------+-----------+-----------------+\n",
            "| [0.0,1.0]|          1|3.918252893250348|\n",
            "| [0.0,1.0]|          1|3.918252893250348|\n",
            "| [0.0,1.0]|          1|3.918252893250348|\n",
            "| [0.0,1.0]|          1|3.918252893250348|\n",
            "| [0.0,1.0]|          1|3.918252893250348|\n",
            "+----------+-----------+-----------------+\n",
            "only showing top 5 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "q7m09KnunW9e",
        "colab_type": "code",
        "outputId": "46fa949f-a5db-4f4a-841a-94f5843c5af2",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 51
        }
      },
      "source": [
        "#coefficient of the regression model\n",
        "coeff = regressor.coefficients\n",
        "#X and Y intercept\n",
        "intr = regressor.intercept\n",
        "print (\"The coefficient of the model is : %a\" %coeff)\n",
        "print (\"The Intercept of the model is : %f\" %intr)"
      ],
      "execution_count": 0,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "The coefficient of the model is : DenseVector([0.097, -0.0914])\n",
            "The Intercept of the model is : 4.007358\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "C3gK6ULloSQd",
        "colab_type": "code",
        "outputId": "4664be51-3aa1-4554-8732-22f4638dc66f",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 85
        }
      },
      "source": [
        "from pyspark.ml.evaluation import RegressionEvaluator\n",
        "eval = RegressionEvaluator(labelCol=\"star_rating\", predictionCol=\"prediction\", metricName=\"rmse\")\n",
        "\n",
        "# Root Mean Square Error\n",
        "rmse = eval.evaluate(predicting.predictions)\n",
        "print(\"RMSE: %.3f\" % rmse)\n",
        "\n",
        "# Mean Square Error\n",
        "mse = eval.evaluate(predicting.predictions, {eval.metricName: \"mse\"})\n",
        "print(\"MSE: %.3f\" % mse)\n",
        "\n",
        "# Mean Absolute Error\n",
        "mae = eval.evaluate(predicting.predictions, {eval.metricName: \"mae\"})\n",
        "print(\"MAE: %.3f\" % mae)\n",
        "\n",
        "# r2 - coefficient of determination\n",
        "r2 = eval.evaluate(predicting.predictions, {eval.metricName: \"r2\"})\n",
        "print(\"r2: %.3f\" %r2)\n"
      ],
      "execution_count": 0,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "RMSE: 1.455\n",
            "MSE: 2.118\n",
            "MAE: 1.210\n",
            "r2: 0.026\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "AsAlvpxgrTx1",
        "colab_type": "text"
      },
      "source": [
        "# Logistic Regression"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "rEyE2dCSxEPU",
        "colab_type": "text"
      },
      "source": [
        "Logistic regression is a statistical method for analyzing a dataset in which there are 'helpful_votes', 'total_votes','star_rating' variables that determine an outcome ('vine' column)"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Upi7rwits9se",
        "colab_type": "code",
        "outputId": "a6b9a596-5236-43b7-9fba-0eb3a374e327",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 204
        }
      },
      "source": [
        "# convert vine column to dichotomous types\n",
        "from pyspark.ml.feature import StringIndexer\n",
        "from pyspark.ml.classification import LogisticRegression\n",
        "from pyspark.ml import Pipeline\n",
        "\n",
        "strIndexed = StringIndexer(inputCol='vine', outputCol='label')\n",
        "\n",
        "assembler = VectorAssembler(inputCols=['helpful_votes', 'total_votes','star_rating'], outputCol='Attributes')\n",
        "\n",
        "data_prep_pipeline = Pipeline(stages= [strIndexed, assembler])\n",
        "\n",
        "pipeline = data_prep_pipeline.fit(vine_reviews_df)\n",
        "lr_df = pipeline.transform(vine_reviews_df)\n",
        "clean_df = lr_df.select(\"Attributes\",\"label\")\n",
        "clean_df.show(5)"
      ],
      "execution_count": 0,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+-------------+-----+\n",
            "|   Attributes|label|\n",
            "+-------------+-----+\n",
            "|[4.0,4.0,1.0]|  0.0|\n",
            "|[1.0,1.0,2.0]|  0.0|\n",
            "|[0.0,1.0,3.0]|  0.0|\n",
            "|[0.0,1.0,3.0]|  0.0|\n",
            "|[3.0,3.0,5.0]|  0.0|\n",
            "+-------------+-----+\n",
            "only showing top 5 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ngj-Q1PRraNr",
        "colab_type": "code",
        "outputId": "8733ab61-c9c3-4e50-dbef-3592ddc7a21c",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 289
        }
      },
      "source": [
        "#Split training and testing data\n",
        "train,test = clean_df.randomSplit([0.7,0.3])\n",
        "\n",
        "lr = LogisticRegression(labelCol =\"label\", featuresCol=\"Attributes\")\n",
        "model=lr.fit(train)\n",
        "predict_train=model.transform(train)\n",
        "predict_test=model.transform(test)\n",
        "predict_test.show(10)"
      ],
      "execution_count": 0,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+-------------+-----+--------------------+--------------------+----------+\n",
            "|   Attributes|label|       rawPrediction|         probability|prediction|\n",
            "+-------------+-----+--------------------+--------------------+----------+\n",
            "|[0.0,1.0,1.0]|  0.0|[4.92895245807518...|[0.99281787870779...|       0.0|\n",
            "|[0.0,1.0,1.0]|  0.0|[4.92895245807518...|[0.99281787870779...|       0.0|\n",
            "|[0.0,1.0,1.0]|  0.0|[4.92895245807518...|[0.99281787870779...|       0.0|\n",
            "|[0.0,1.0,1.0]|  0.0|[4.92895245807518...|[0.99281787870779...|       0.0|\n",
            "|[0.0,1.0,1.0]|  0.0|[4.92895245807518...|[0.99281787870779...|       0.0|\n",
            "|[0.0,1.0,1.0]|  0.0|[4.92895245807518...|[0.99281787870779...|       0.0|\n",
            "|[0.0,1.0,1.0]|  0.0|[4.92895245807518...|[0.99281787870779...|       0.0|\n",
            "|[0.0,1.0,1.0]|  0.0|[4.92895245807518...|[0.99281787870779...|       0.0|\n",
            "|[0.0,1.0,1.0]|  0.0|[4.92895245807518...|[0.99281787870779...|       0.0|\n",
            "|[0.0,1.0,1.0]|  0.0|[4.92895245807518...|[0.99281787870779...|       0.0|\n",
            "+-------------+-----+--------------------+--------------------+----------+\n",
            "only showing top 10 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "tWHDeWcKwZcs",
        "colab_type": "code",
        "outputId": "e1408e85-c9ec-48e4-c501-18f3613ee865",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 238
        }
      },
      "source": [
        "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
        "evaluator=BinaryClassificationEvaluator(rawPredictionCol='rawPrediction',labelCol='label')\n",
        "predict_test.select(\"label\",\"rawPrediction\",\"prediction\",\"probability\").show(5)\n",
        "print(\"The area under ROC for train set is {}\".format(evaluator.evaluate(predict_train)))\n",
        "print(\"The area under ROC for test set is {}\".format(evaluator.evaluate(predict_test)))"
      ],
      "execution_count": 0,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+-----+--------------------+----------+--------------------+\n",
            "|label|       rawPrediction|prediction|         probability|\n",
            "+-----+--------------------+----------+--------------------+\n",
            "|  0.0|[4.92895245807518...|       0.0|[0.99281787870779...|\n",
            "|  0.0|[4.92895245807518...|       0.0|[0.99281787870779...|\n",
            "|  0.0|[4.92895245807518...|       0.0|[0.99281787870779...|\n",
            "|  0.0|[4.92895245807518...|       0.0|[0.99281787870779...|\n",
            "|  0.0|[4.92895245807518...|       0.0|[0.99281787870779...|\n",
            "+-----+--------------------+----------+--------------------+\n",
            "only showing top 5 rows\n",
            "\n",
            "The area under ROC for train set is 0.5978297406156183\n",
            "The area under ROC for test set is 0.6018163911448003\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "SY7rGWXHxT18",
        "colab_type": "text"
      },
      "source": [
        "The area under the ROC curve (AUC) is a measure of how well parameters can distinguish between two groups. In this case, Predict AUC is 0.6 means the threshold would be 60%. "
      ]
    }
  ]
}