Amazon_Reviews_ETL.ipynb

{
  "cells": [
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "V58rxea0HqSa",
        "outputId": "6b6c2158-5a99-43f2-d34e-24a66a4e8db6"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "\r0% [Working]\r            \rGet:1 https://cloud.r-project.org/bin/linux/ubuntu bionic-cran40/ InRelease [3,626 B]\n",
            "\r0% [Connecting to archive.ubuntu.com] [Connecting to security.ubuntu.com (185.1\r0% [Connecting to archive.ubuntu.com] [Connecting to security.ubuntu.com (185.1\r0% [1 InRelease gpgv 3,626 B] [Connecting to archive.ubuntu.com] [Connecting to\r                                                                               \rIgn:2 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  InRelease\n",
            "\r0% [1 InRelease gpgv 3,626 B] [Connecting to archive.ubuntu.com] [Connecting to\r                                                                               \rHit:3 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  InRelease\n",
            "Hit:4 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  Release\n",
            "Hit:5 http://archive.ubuntu.com/ubuntu bionic InRelease\n",
            "Get:6 http://security.ubuntu.com/ubuntu bionic-security InRelease [88.7 kB]\n",
            "Hit:7 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic InRelease\n",
            "Get:8 http://archive.ubuntu.com/ubuntu bionic-updates InRelease [88.7 kB]\n",
            "Hit:9 http://ppa.launchpad.net/cran/libgit2/ubuntu bionic InRelease\n",
            "Get:10 http://archive.ubuntu.com/ubuntu bionic-backports InRelease [83.3 kB]\n",
            "Hit:11 http://ppa.launchpad.net/deadsnakes/ppa/ubuntu bionic InRelease\n",
            "Hit:13 http://ppa.launchpad.net/graphics-drivers/ppa/ubuntu bionic InRelease\n",
            "Get:14 http://archive.ubuntu.com/ubuntu bionic-updates/restricted amd64 Packages [1,300 kB]\n",
            "Get:15 http://archive.ubuntu.com/ubuntu bionic-updates/universe amd64 Packages [2,336 kB]\n",
            "Get:16 http://archive.ubuntu.com/ubuntu bionic-updates/multiverse amd64 Packages [29.8 kB]\n",
            "Get:17 http://archive.ubuntu.com/ubuntu bionic-updates/main amd64 Packages [3,492 kB]\n",
            "Get:18 http://archive.ubuntu.com/ubuntu bionic-backports/universe amd64 Packages [23.9 kB]\n",
            "Get:19 http://archive.ubuntu.com/ubuntu bionic-backports/main amd64 Packages [93.6 kB]\n",
            "Get:20 http://security.ubuntu.com/ubuntu bionic-security/main amd64 Packages [3,066 kB]\n",
            "Get:21 http://security.ubuntu.com/ubuntu bionic-security/restricted amd64 Packages [1,259 kB]\n",
            "Get:22 http://security.ubuntu.com/ubuntu bionic-security/universe amd64 Packages [1,561 kB]\n",
            "Fetched 13.4 MB in 4s (3,258 kB/s)\n",
            "Reading package lists... Done\n"
          ]
        }
      ],
      "source": [
        "import os\n",
        "# Find the latest version of spark 3.0  from http://www.apache.org/dist/spark/ and enter as the spark version\n",
        "# For example:\n",
        "# spark_version = 'spark-3.0.3'\n",
        "spark_version = 'spark-3.3.1'\n",
        "os.environ['SPARK_VERSION']=spark_version\n",
        "\n",
        "# Install Spark and Java\n",
        "!apt-get update\n",
        "!apt-get install openjdk-11-jdk-headless -qq > /dev/null\n",
        "!wget -q http://www.apache.org/dist/spark/$SPARK_VERSION/$SPARK_VERSION-bin-hadoop3.tgz\n",
        "!tar xf $SPARK_VERSION-bin-hadoop3.tgz\n",
        "!pip install -q findspark\n",
        "\n",
        "# Set Environment Variables\n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-11-openjdk-amd64\"\n",
        "os.environ[\"SPARK_HOME\"] = f\"/content/{spark_version}-bin-hadoop3\"\n",
        "\n",
        "# Start a SparkSession\n",
        "import findspark\n",
        "findspark.init()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "_xKwTpATHqSe",
        "outputId": "93b5ecd3-077a-4542-cd07-6aca961614c8"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "--2022-11-20 17:09:58--  https://jdbc.postgresql.org/download/postgresql-42.2.16.jar\n",
            "Resolving jdbc.postgresql.org (jdbc.postgresql.org)... 72.32.157.228, 2001:4800:3e1:1::228\n",
            "Connecting to jdbc.postgresql.org (jdbc.postgresql.org)|72.32.157.228|:443... connected.\n",
            "HTTP request sent, awaiting response... 200 OK\n",
            "Length: 1002883 (979K) [application/java-archive]\n",
            "Saving to: ‘postgresql-42.2.16.jar’\n",
            "\n",
            "postgresql-42.2.16. 100%[===================>] 979.38K  4.94MB/s    in 0.2s    \n",
            "\n",
            "2022-11-20 17:09:59 (4.94 MB/s) - ‘postgresql-42.2.16.jar’ saved [1002883/1002883]\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# Download the Postgres driver that will allow Spark to interact with Postgres.\n",
        "!wget https://jdbc.postgresql.org/download/postgresql-42.2.16.jar"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {
        "id": "MMqDAjVS0KN9"
      },
      "outputs": [],
      "source": [
        "from pyspark.sql import SparkSession\n",
        "spark = SparkSession.builder.appName(\"M16-Amazon-Challenge\").config(\"spark.driver.extraClassPath\",\"/content/postgresql-42.2.16.jar\").getOrCreate()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cyBsySGuY-9V"
      },
      "source": [
        "### Load Amazon Data into Spark DataFrame"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 4,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "CtCmBhQJY-9Z",
        "outputId": "8e4d7f4d-05a5-454d-cfe4-645873dbb471"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-----------+-----------+--------------+----------+--------------+--------------------+----------------+-----------+-------------+-----------+----+-----------------+--------------------+--------------------+-------------------+\n",
            "|marketplace|customer_id|     review_id|product_id|product_parent|       product_title|product_category|star_rating|helpful_votes|total_votes|vine|verified_purchase|     review_headline|         review_body|        review_date|\n",
            "+-----------+-----------+--------------+----------+--------------+--------------------+----------------+-----------+-------------+-----------+----+-----------------+--------------------+--------------------+-------------------+\n",
            "|         US|   48945260|R1WBPB8MDCCN8F|B012P7UPSM|     409940130|Chicago Blackhawk...|          Sports|          5|            0|          0|   N|                N|   LOVE IT. 6 stars!|Bought this last ...|2015-08-31 00:00:00|\n",
            "|         US|    5782091|R32M0YEWV77XG8|B001GQ3VHG|     657746679|Copag Poker Size ...|          Sports|          5|            1|          1|   N|                Y|       Shipped fast.|These are the bes...|2015-08-31 00:00:00|\n",
            "|         US|   45813853| RR8V7WR27NXJ5|B008VS8M58|     962442336|Baoer 223 5.56x45...|          Sports|          1|            0|          0|   N|                Y|Good idea if it w...|It looks good, th...|2015-08-31 00:00:00|\n",
            "|         US|    1593730|R1MHO5V9Z932AY|B005F06F4U|      74305227|All Terrain Tackl...|          Sports|          5|            0|          0|   N|                Y|          Five Stars|           Great jig|2015-08-31 00:00:00|\n",
            "|         US|   29605511|R16PD71086BD2V|B010T4IE2C|     787185588|Swim Cap - 3 Pack...|          Sports|          5|            0|          1|   N|                N|Great quality sil...|I love swimming i...|2015-08-31 00:00:00|\n",
            "|         US|   11112959|R1Z8IFGWTRWXT6|B004RKJGLS|      94127483|adidas Men's Spor...|          Sports|          3|            0|          0|   N|                Y|Love everything a...|Love everything a...|2015-08-31 00:00:00|\n",
            "|         US|     108031|R3AUMSHAW73HWN|B005V3DCBU|     526977496|Nike Men's Perfor...|          Sports|          4|            2|          3|   N|                N|          Four Stars|not the best sock...|2015-08-31 00:00:00|\n",
            "|         US|   13981540|R2KWDWFOHGX6FL|B00MHT9WN8|      26521265|Green Bay Packers...|          Sports|          5|            0|          0|   N|                Y|          Five Stars|             Love it|2015-08-31 00:00:00|\n",
            "|         US|   37993909|R3H9543FWBWFBU|B001CSIRQ8|     652431165|Isokinetics Inc. ...|          Sports|          5|            1|          1|   N|                Y|can't wait to use it|Unfortunately I h...|2015-08-31 00:00:00|\n",
            "|         US|   26040213| RUANXOQ9W3OU5|B001KZ3NOO|     635861713|Aottop Quality El...|          Sports|          5|            0|          0|   N|                Y|great product, ac...|fast shipping , g...|2015-08-31 00:00:00|\n",
            "|         US|   34657602|R31673RTGEZSW7|B00005RCQS|      72099763|Everlast 2'x6' Fo...|          Sports|          5|            2|          2|   N|                Y|this mat is a pre...|If you have a ter...|2015-08-31 00:00:00|\n",
            "|         US|   14346192|R22OQLFSH42RCM|B00FA7RWVI|     757354022|OGIO Men's Shredd...|          Sports|          5|            1|          1|   N|                Y|  Versatile Golf Bag|Love this golf ba...|2015-08-31 00:00:00|\n",
            "|         US|   38782687|R12LEL4F3TSZUJ|B000MMH2HU|     272402599|Blue Sea Systems ...|          Sports|          5|            2|          2|   N|                Y|Great AC Main Bre...|Perfect mounting ...|2015-08-31 00:00:00|\n",
            "|         US|   27138575|R2L9XWD03072NI|B00WJOATDS|     375070208|Zengi AR15 45 Deg...|          Sports|          5|            1|          2|   N|                Y|   Nice, some quirks|For the cost you ...|2015-08-31 00:00:00|\n",
            "|         US|   11838771|R2K0U91HIACANO|B00BOM2JNY|     639666785|Black Mountain Pr...|          Sports|          5|            0|          0|   N|                N|55 cm - Good qual...|The ball is very ...|2015-08-31 00:00:00|\n",
            "|         US|     535800|R29SP6MBT7MXG7|B00OD5GAFM|      38715442|Trijicon VCOG 1-6...|          Sports|          4|            3|          3|   N|                Y|    Very nice scope.|Very nice scope.....|2015-08-31 00:00:00|\n",
            "|         US|   23156579|R31XREAAMATEPY|B00CAHDC1K|      57088652|Naruto Headband f...|          Sports|          5|            0|          0|   N|                Y|was for my son an...|           This item|2015-08-31 00:00:00|\n",
            "|         US|   48107879|R2116AVB87SO38|B004NLHXLG|     114125984|Under Armour Men'...|          Sports|          5|            0|          0|   N|                Y|          Five Stars|greatest socks i ...|2015-08-31 00:00:00|\n",
            "|         US|   27260960|R3RDVBB6O0X3HW|B00L7OANWI|     883962979|Yes4All Deep Tiss...|          Sports|          5|            0|          0|   N|                Y|          Five Stars|Perfect for home ...|2015-08-31 00:00:00|\n",
            "|         US|   39537314|R3LW5T149LKKQM|B0085PPSIQ|     691479969|Lansky PS-MED01 B...|          Sports|          4|            0|          0|   N|                Y|A bit heavy for p...|Very good product...|2015-08-31 00:00:00|\n",
            "+-----------+-----------+--------------+----------+--------------+--------------------+----------------+-----------+-------------+-----------+----+-----------------+--------------------+--------------------+-------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "from pyspark import SparkFiles\n",
        "url = \"https://s3.amazonaws.com/amazon-reviews-pds/tsv/amazon_reviews_us_Sports_v1_00.tsv.gz\"\n",
        "spark.sparkContext.addFile(url)\n",
        "df = spark.read.option(\"encoding\", \"UTF-8\").csv(SparkFiles.get(\"\"), sep=\"\\t\", header=True, inferSchema=True)\n",
        "df.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "2yUSe55VY-9t"
      },
      "source": [
        "### Create DataFrames to match tables"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 5,
      "metadata": {
        "id": "C8REmY1aY-9u"
      },
      "outputs": [],
      "source": [
        "from pyspark.sql.functions import to_date\n",
        "# Read in the Review dataset as a DataFrame\n",
        "\n",
        "# PySpark groupBy() agg\n",
        "from pyspark.sql.functions import count"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 6,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "B0TESUDRY-90",
        "outputId": "c31e3bc7-dbb2-4f28-e0e7-b24c66935486"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-----------+--------------+\n",
            "|customer_id|customer_count|\n",
            "+-----------+--------------+\n",
            "|   35669025|             1|\n",
            "|   48198949|            30|\n",
            "|   43355824|             6|\n",
            "|   33014261|             6|\n",
            "|   23493243|             1|\n",
            "+-----------+--------------+\n",
            "only showing top 5 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# Create the customers_table DataFrame\n",
        "customers_df = df.groupby(\"customer_id\").agg(count(\"customer_id\")).withColumnRenamed(\"count(customer_id)\", \"customer_count\")\n",
        "customers_df.show(5)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 7,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "4FwXA6UvY-96",
        "outputId": "97d69409-63a8-4bcf-8ff6-4a7ba6cdb3e7"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----------+--------------------+\n",
            "|product_id|       product_title|\n",
            "+----------+--------------------+\n",
            "|0000031852|Girls Ballet Tutu...|\n",
            "|0000031909|Mystiqueshapes Gi...|\n",
            "|0201593246|Official Arsenal ...|\n",
            "|0395911737|Peterson Field Gu...|\n",
            "|052800607X|Rand McNally Inte...|\n",
            "+----------+--------------------+\n",
            "only showing top 5 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# Create the products_table DataFrame and drop duplicates. \n",
        "products_df = df.select([\"product_id\",\"product_title\"]).drop_duplicates([\"product_id\"])\n",
        "products_df.show(5)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 8,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "MkqyCuNQY-9-",
        "outputId": "77033a5d-d3f9-426d-cab9-d9b857b7683c"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------------+-----------+----------+--------------+-----------+\n",
            "|     review_id|customer_id|product_id|product_parent|review_date|\n",
            "+--------------+-----------+----------+--------------+-----------+\n",
            "|R1WBPB8MDCCN8F|   48945260|B012P7UPSM|     409940130| 2015-08-31|\n",
            "|R32M0YEWV77XG8|    5782091|B001GQ3VHG|     657746679| 2015-08-31|\n",
            "| RR8V7WR27NXJ5|   45813853|B008VS8M58|     962442336| 2015-08-31|\n",
            "|R1MHO5V9Z932AY|    1593730|B005F06F4U|      74305227| 2015-08-31|\n",
            "|R16PD71086BD2V|   29605511|B010T4IE2C|     787185588| 2015-08-31|\n",
            "+--------------+-----------+----------+--------------+-----------+\n",
            "only showing top 5 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# Create the review_id_table DataFrame. \n",
        "# Convert the 'review_date' column to a date datatype with to_date(\"review_date\", 'yyyy-MM-dd').alias(\"review_date\")\n",
        "review_id_df = df.select([\"review_id\", \"customer_id\", \"product_id\",\"product_parent\", to_date(\"review_date\", 'yyyy-MM-dd').alias(\"review_date\")])\n",
        "review_id_df.show(5)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 9,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "lzMmkdKmY--D",
        "outputId": "dbb52e23-25f8-48e9-d9ee-08cb46463da2"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------------+-----------+-------------+-----------+----+-----------------+\n",
            "|     review_id|star_rating|helpful_votes|total_votes|vine|verified_purchase|\n",
            "+--------------+-----------+-------------+-----------+----+-----------------+\n",
            "|R1WBPB8MDCCN8F|          5|            0|          0|   N|                N|\n",
            "|R32M0YEWV77XG8|          5|            1|          1|   N|                Y|\n",
            "| RR8V7WR27NXJ5|          1|            0|          0|   N|                Y|\n",
            "|R1MHO5V9Z932AY|          5|            0|          0|   N|                Y|\n",
            "|R16PD71086BD2V|          5|            0|          1|   N|                N|\n",
            "+--------------+-----------+-------------+-----------+----+-----------------+\n",
            "only showing top 5 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# Create the vine_table. DataFrame\n",
        "vine_df = df.select([\"review_id\", \"star_rating\",\"helpful_votes\",\"total_votes\",\"vine\", \"verified_purchase\"])\n",
        "# convert star rating from a character to a number\n",
        "vine_df.show(5)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "jITZhLkmY--J"
      },
      "source": [
        "### Connect to the AWS RDS instance and write each DataFrame to its table. "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 10,
      "metadata": {
        "id": "7jiUvs1aY--L"
      },
      "outputs": [],
      "source": [
        "# Configure settings for RDS\n",
        "mode = \"append\"\n",
        "jdbc_url=\"jdbc:postgresql://deliv1.cttm8dkkyji9.us-east-1.rds.amazonaws.com:5432/postgres\"\n",
        "config = {\"user\":\"postgres\", \n",
        "          \"password\": \"Atl1025!\", \n",
        "          \"driver\":\"org.postgresql.Driver\"}"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 12,
      "metadata": {
        "id": "T2zgZ-aKY--Q"
      },
      "outputs": [],
      "source": [
        "# Write review_id_df to table in RDS\n",
        "review_id_df.write.jdbc(url=jdbc_url, table='review_id_table', mode=mode, properties=config)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 13,
      "metadata": {
        "id": "1m3yzn-LY--U"
      },
      "outputs": [],
      "source": [
        "# Write products_df to table in RDS\n",
        "# about 3 min\n",
        "\n",
        "products_df.write.jdbc(url=jdbc_url, table='products_table', mode=mode, properties=config)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 18,
      "metadata": {
        "id": "KbXri15fY--Z"
      },
      "outputs": [],
      "source": [
        "# Write customers_df to table in RDS\n",
        "# 5 min 14 s\n",
        "# actually took 7 minutes\n",
        "customers_df.write.jdbc(url=jdbc_url, table='customers_table', mode=mode, properties=config)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 15,
      "metadata": {
        "id": "w3a-eudVc3d9"
      },
      "outputs": [],
      "source": [
        "from pyspark.sql.types import IntegerType"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 16,
      "metadata": {
        "id": "NLA3bhabdAAW"
      },
      "outputs": [],
      "source": [
        "vine_df = vine_df.withColumn(\"star_rating\",vine_df.star_rating.cast(IntegerType()))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 17,
      "metadata": {
        "id": "XdQknSHLY--e"
      },
      "outputs": [],
      "source": [
        "# Write vine_df to table in RDS\n",
        "# 11 minutes\n",
        "# actually took 23 minutes\n",
        "\n",
        "vine_df.write.jdbc(url=jdbc_url, table='vine_table', mode=mode, properties=config)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Exuo6ebUsCqW"
      },
      "outputs": [],
      "source": []
    }
  ],
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    },
    "nteract": {
      "version": "0.12.3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}