Amazon_Reviews_ETL.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "V58rxea0HqSa",
    "outputId": "f61a25e0-8ae7-4e13-8f0b-b3f7c61ec022"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\r",
      "0% [Working]\r",
      "            \r",
      "Get:1 https://cloud.r-project.org/bin/linux/ubuntu bionic-cran40/ InRelease [3,626 B]\n",
      "\r",
      "0% [Connecting to archive.ubuntu.com (91.189.88.152)] [Connecting to security.u\r",
      "0% [Connecting to archive.ubuntu.com (91.189.88.152)] [Connecting to security.u\r",
      "0% [1 InRelease gpgv 3,626 B] [Connecting to archive.ubuntu.com (91.189.88.152)\r",
      "                                                                               \r",
      "Ign:2 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  InRelease\n",
      "\r",
      "0% [1 InRelease gpgv 3,626 B] [Waiting for headers] [Waiting for headers] [Wait\r",
      "                                                                               \r",
      "Hit:3 http://archive.ubuntu.com/ubuntu bionic InRelease\n",
      "\r",
      "0% [1 InRelease gpgv 3,626 B] [Waiting for headers] [Waiting for headers] [Wait\r",
      "                                                                               \r",
      "Get:4 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic InRelease [15.9 kB]\n",
      "\r",
      "0% [1 InRelease gpgv 3,626 B] [Waiting for headers] [Waiting for headers] [4 In\r",
      "                                                                               \r",
      "Get:5 http://security.ubuntu.com/ubuntu bionic-security InRelease [88.7 kB]\n",
      "\r",
      "0% [1 InRelease gpgv 3,626 B] [Waiting for headers] [5 InRelease 14.2 kB/88.7 k\r",
      "                                                                               \r",
      "Ign:6 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  InRelease\n",
      "\r",
      "0% [1 InRelease gpgv 3,626 B] [Waiting for headers] [5 InRelease 14.2 kB/88.7 k\r",
      "                                                                               \r",
      "Hit:7 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  Release\n",
      "Hit:8 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  Release\n",
      "Get:9 http://archive.ubuntu.com/ubuntu bionic-updates InRelease [88.7 kB]\n",
      "Get:10 https://cloud.r-project.org/bin/linux/ubuntu bionic-cran40/ Packages [49.8 kB]\n",
      "Hit:11 http://ppa.launchpad.net/cran/libgit2/ubuntu bionic InRelease\n",
      "Get:12 http://archive.ubuntu.com/ubuntu bionic-backports InRelease [74.6 kB]\n",
      "Hit:13 http://ppa.launchpad.net/deadsnakes/ppa/ubuntu bionic InRelease\n",
      "Get:14 http://ppa.launchpad.net/graphics-drivers/ppa/ubuntu bionic InRelease [21.3 kB]\n",
      "Get:17 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic/main Sources [1,749 kB]\n",
      "Get:18 http://security.ubuntu.com/ubuntu bionic-security/universe amd64 Packages [1,402 kB]\n",
      "Get:19 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic/main amd64 Packages [895 kB]\n",
      "Get:20 http://archive.ubuntu.com/ubuntu bionic-updates/universe amd64 Packages [2,170 kB]\n",
      "Get:21 http://security.ubuntu.com/ubuntu bionic-security/main amd64 Packages [2,045 kB]\n",
      "Get:22 http://security.ubuntu.com/ubuntu bionic-security/restricted amd64 Packages [348 kB]\n",
      "Get:23 http://archive.ubuntu.com/ubuntu bionic-updates/restricted amd64 Packages [378 kB]\n",
      "Get:24 http://archive.ubuntu.com/ubuntu bionic-updates/main amd64 Packages [2,475 kB]\n",
      "Get:25 http://ppa.launchpad.net/graphics-drivers/ppa/ubuntu bionic/main amd64 Packages [49.4 kB]\n",
      "Fetched 11.9 MB in 4s (3,331 kB/s)\n",
      "Reading package lists... Done\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "# Find the latest version of spark 2.0  from http://www-us.apache.org/dist/spark/ and enter as the spark version\n",
    "# For example:\n",
    "# spark_version = 'spark-3.0.0'\n",
    "spark_version = 'spark-3.0.2'\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-us.apache.org/dist/spark/$SPARK_VERSION/$SPARK_VERSION-bin-hadoop2.7.tgz\n",
    "!tar xf $SPARK_VERSION-bin-hadoop2.7.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-hadoop2.7\"\n",
    "\n",
    "# Start a SparkSession\n",
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "_xKwTpATHqSe",
    "outputId": "a4d94d4b-d1a5-4715-caa8-26e2d2f228e3"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--2021-03-30 16:26:22--  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  5.69MB/s    in 0.2s    \n",
      "\n",
      "2021-03-30 16:26:23 (5.69 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": null,
   "metadata": {
    "id": "MMqDAjVS0KN9"
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName(\"BigData-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": null,
   "metadata": {
    "id": "CtCmBhQJY-9Z"
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkFiles\n",
    "url = \"https://s3.amazonaws.com/amazon-reviews-pds/tsv/amazon_reviews_us_Video_Games_v1_00.tsv.gz\"\n",
    "spark.sparkContext.addFile(url)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "2yUSe55VY-9t"
   },
   "source": [
    "### Create DataFrames to match tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "C8REmY1aY-9u",
    "outputId": "5abc974c-f944-458e-8904-63c6e312fb04"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "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|   12039526| RTIS3L2M1F5SM|B001CXYMFS|     737716809|Thrustmaster T-Fl...|     Video Games|          5|            0|          0|   N|                Y|an amazing joysti...|Used this for Eli...| 2015-08-31|\n",
      "|         US|    9636577| R1ZV7R40OLHKD|B00M920ND6|     569686175|Tonsee 6 buttons ...|     Video Games|          5|            0|          0|   N|                Y|Definitely a sile...|Loved it,  I didn...| 2015-08-31|\n",
      "|         US|    2331478|R3BH071QLH8QMC|B0029CSOD2|      98937668|Hidden Mysteries:...|     Video Games|          1|            0|          1|   N|                Y|            One Star|poor quality work...| 2015-08-31|\n",
      "|         US|   52495923|R127K9NTSXA2YH|B00GOOSV98|      23143350|GelTabz Performan...|     Video Games|          3|            0|          0|   N|                Y|good, but could b...|nice, but tend to...| 2015-08-31|\n",
      "|         US|   14533949|R32ZWUXDJPW27Q|B00Y074JOM|     821342511|Zero Suit Samus a...|     Video Games|          4|            0|          0|   N|                Y|   Great but flawed.|Great amiibo, gre...| 2015-08-31|\n",
      "+-----------+-----------+--------------+----------+--------------+--------------------+----------------+-----------+-------------+-----------+----+-----------------+--------------------+--------------------+-----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import to_date\n",
    "from pyspark.sql.functions import count\n",
    "# Read in the Review dataset as a DataFrame\n",
    "df = spark.read.option(\"encoding\", \"UTF-8\").csv(SparkFiles.get(\"\"), sep=\"\\t\", header=True, inferSchema=True)\n",
    "df.show(5)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "B0TESUDRY-90",
    "outputId": "d66acb9a-61ff-4887-935d-b879062c5b3f"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------------+\n",
      "|customer_id|customer_count|\n",
      "+-----------+--------------+\n",
      "|   48670265|             1|\n",
      "|   49103216|             2|\n",
      "|    1131200|             1|\n",
      "|   43076447|             2|\n",
      "|   46261368|             1|\n",
      "+-----------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Create the customers_table DataFrame\n",
    "customers_df = df.groupby(\"customer_id\").agg({\"customer_id\":\"count\"}).withColumnRenamed(\"count(customer_id)\", \"customer_count\")\n",
    "customers_df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "4FwXA6UvY-96",
    "outputId": "43aa59c2-ad69-4ea9-b137-38bbafe0f5cf"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+\n",
      "|product_id|       product_title|\n",
      "+----------+--------------------+\n",
      "|B00CJ7IUI6|The Elder Scrolls...|\n",
      "|B00DHF39KS|Wolfenstein: The ...|\n",
      "|B00MUTAVH6|Under Night In-Bi...|\n",
      "|B001AZSEUW|              Peggle|\n",
      "|B00KVOVBGM|PlayStation 4 Con...|\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()\n",
    "products_df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "MkqyCuNQY-9-",
    "outputId": "ee920530-5e7f-4783-8077-8175374686ea"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+-----------+----------+--------------+-----------+\n",
      "|     review_id|customer_id|product_id|product_parent|review_date|\n",
      "+--------------+-----------+----------+--------------+-----------+\n",
      "| RTIS3L2M1F5SM|   12039526|B001CXYMFS|     737716809| 2015-08-31|\n",
      "| R1ZV7R40OLHKD|    9636577|B00M920ND6|     569686175| 2015-08-31|\n",
      "|R3BH071QLH8QMC|    2331478|B0029CSOD2|      98937668| 2015-08-31|\n",
      "|R127K9NTSXA2YH|   52495923|B00GOOSV98|      23143350| 2015-08-31|\n",
      "|R32ZWUXDJPW27Q|   14533949|B00Y074JOM|     821342511| 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": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "lzMmkdKmY--D",
    "outputId": "45b769bf-bc3a-471b-f975-e20d2c6b9ce0"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+-----------+-------------+-----------+----+-----------------+\n",
      "|     review_id|star_rating|helpful_votes|total_votes|vine|verified_purchase|\n",
      "+--------------+-----------+-------------+-----------+----+-----------------+\n",
      "| RTIS3L2M1F5SM|          5|            0|          0|   N|                Y|\n",
      "| R1ZV7R40OLHKD|          5|            0|          0|   N|                Y|\n",
      "|R3BH071QLH8QMC|          1|            0|          1|   N|                Y|\n",
      "|R127K9NTSXA2YH|          3|            0|          0|   N|                Y|\n",
      "|R32ZWUXDJPW27Q|          4|            0|          0|   N|                Y|\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",
    "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": null,
   "metadata": {
    "id": "7jiUvs1aY--L"
   },
   "outputs": [],
   "source": [
    "# Configure settings for RDS\n",
    "mode = \"append\"\n",
    "jdbc_url=\"jdbc:postgresql://external-data.cwztyazhcrmp.us-east-2.rds.amazonaws.com:5432/postgres\"\n",
    "config = {\"user\":\"postgres\", \n",
    "          \"password\": \"brillas1986\", \n",
    "          \"driver\":\"org.postgresql.Driver\"}\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "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": null,
   "metadata": {
    "id": "1m3yzn-LY--U"
   },
   "outputs": [],
   "source": [
    "# Write products_df to table in RDS\n",
    "# about 3 min\n",
    "products_df.write.jdbc(url=jdbc_url, table='products_table', mode=mode, properties=config)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "KbXri15fY--Z"
   },
   "outputs": [],
   "source": [
    "# Write customers_df to table in RDS\n",
    "# 5 min 14 s\n",
    "customers_df.write.jdbc(url=jdbc_url, table='customers_table', mode=mode, properties=config)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "XdQknSHLY--e"
   },
   "outputs": [],
   "source": [
    "# Write vine_df to table in RDS\n",
    "# 11 minutes\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": {
   "name": "Amazon_Reviews_ETL.ipynb",
   "provenance": []
  },
  "kernelspec": {
   "display_name": "PythonData",
   "language": "python",
   "name": "pythondata"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.9"
  },
  "nteract": {
   "version": "0.12.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}