Amazon_Reviews_ETL.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Amazon_Reviews_ETL.ipynb",
      "provenance": [],
      "collapsed_sections": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "CWl_IYgty-5s",
        "outputId": "d680a6bc-fcf4-4b4f-d05d-1ddc6bd0d88b"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "\r0% [Working]\r            \rHit:1 http://archive.ubuntu.com/ubuntu bionic InRelease\n",
            "\r0% [Connecting to security.ubuntu.com (91.189.91.39)] [Connecting to cloud.r-pr\r                                                                               \rGet:2 http://archive.ubuntu.com/ubuntu bionic-updates InRelease [88.7 kB]\n",
            "\r                                                                               \rGet:3 http://archive.ubuntu.com/ubuntu bionic-backports InRelease [74.6 kB]\n",
            "\r                                                                               \rGet:4 https://cloud.r-project.org/bin/linux/ubuntu bionic-cran40/ InRelease [3,626 B]\n",
            "\r                                                                               \rGet:5 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic InRelease [15.9 kB]\n",
            "\r                                                                               \rHit:6 http://ppa.launchpad.net/cran/libgit2/ubuntu bionic InRelease\n",
            "\r                                                                               \rHit:7 http://ppa.launchpad.net/deadsnakes/ppa/ubuntu bionic InRelease\n",
            "\r                                                                               \rHit:8 http://ppa.launchpad.net/graphics-drivers/ppa/ubuntu bionic InRelease\n",
            "Get:9 http://security.ubuntu.com/ubuntu bionic-security InRelease [88.7 kB]\n",
            "Ign:10 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  InRelease\n",
            "Get:11 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  InRelease [1,581 B]\n",
            "Hit:12 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  Release\n",
            "Get:13 http://archive.ubuntu.com/ubuntu bionic-updates/universe amd64 Packages [2,311 kB]\n",
            "Get:14 http://archive.ubuntu.com/ubuntu bionic-updates/main amd64 Packages [3,369 kB]\n",
            "Get:15 https://cloud.r-project.org/bin/linux/ubuntu bionic-cran40/ Packages [91.1 kB]\n",
            "Get:16 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic/main Sources [2,093 kB]\n",
            "Get:17 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic/main amd64 Packages [1,073 kB]\n",
            "Get:18 http://security.ubuntu.com/ubuntu bionic-security/main amd64 Packages [2,938 kB]\n",
            "Get:19 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  Packages [912 kB]\n",
            "Get:20 http://security.ubuntu.com/ubuntu bionic-security/universe amd64 Packages [1,534 kB]\n",
            "Fetched 14.6 MB in 5s (2,705 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.0'\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",
      "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"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "RW6yr7TazCBV",
        "outputId": "85a3d513-3f8b-46e8-ca09-ba526b8d72be"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "--2022-08-27 00:56:45--  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  1.59MB/s    in 0.6s    \n",
            "\n",
            "2022-08-27 00:56:46 (1.59 MB/s) - ‘postgresql-42.2.16.jar’ saved [1002883/1002883]\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "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()"
      ],
      "metadata": {
        "id": "OiwjTKgMzR0z"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "### Load Amazon Data into Spark DataFrame"
      ],
      "metadata": {
        "id": "aHRqJw_C49Iy"
      }
    },
    {
      "cell_type": "code",
      "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)\n",
        "df = spark.read.option(\"encoding\", \"UTF-8\").csv(SparkFiles.get(\"amazon_reviews_us_Video_Games_v1_00.tsv.gz\"), sep=\"\\t\", header=True, inferSchema=True)\n",
        "df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "cULtrpAdzWPO",
        "outputId": "81ada621-d834-47f6-a800-3b008fac29a5"
      },
      "execution_count": null,
      "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|   12039526| RTIS3L2M1F5SM|B001CXYMFS|     737716809|Thrustmaster T-Fl...|     Video Games|          5|            0|          0|   N|                Y|an amazing joysti...|Used this for Eli...|2015-08-31 00:00:00|\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 00:00:00|\n",
            "|         US|    2331478|R3BH071QLH8QMC|B0029CSOD2|      98937668|Hidden Mysteries:...|     Video Games|          1|            0|          1|   N|                Y|            One Star|poor quality work...|2015-08-31 00:00:00|\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 00:00:00|\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 00:00:00|\n",
            "|         US|    2377552|R3AQQ4YUKJWBA6|B002UBI6W6|     328764615|Psyclone Recharge...|     Video Games|          1|            0|          0|   N|                Y|            One Star|The remote consta...|2015-08-31 00:00:00|\n",
            "|         US|   17521011|R2F0POU5K6F73F|B008XHCLFO|      24234603|Protection for yo...|     Video Games|          5|            0|          0|   N|                Y|              A Must|I have a 2012-201...|2015-08-31 00:00:00|\n",
            "|         US|   19676307|R3VNR804HYSMR6|B00BRA9R6A|     682267517|   Nerf 3DS XL Armor|     Video Games|          5|            0|          0|   N|                Y|          Five Stars|Perfect, kids lov...|2015-08-31 00:00:00|\n",
            "|         US|     224068| R3GZTM72WA2QH|B009EPWJLA|     435241890|One Piece: Pirate...|     Video Games|          5|            0|          0|   N|                Y|          Five Stars|            Excelent|2015-08-31 00:00:00|\n",
            "|         US|   48467989| RNQOY62705W1K|B0000AV7GB|     256572651|Playstation 2 Dan...|     Video Games|          4|            0|          0|   N|                Y|          Four Stars|Slippery but expe...|2015-08-31 00:00:00|\n",
            "|         US|     106569|R1VTIA3JTYBY02|B00008KTNN|     384411423|Metal Arms: Glitc...|     Video Games|          5|            0|          0|   N|                N|          Five Stars|Love the game. Se...|2015-08-31 00:00:00|\n",
            "|         US|   48269642|R29DOU8791QZL8|B000A3IA0Y|     472622859|72 Pin Connector ...|     Video Games|          1|            0|          0|   N|                Y| Game will get stuck|Does not fit prop...|2015-08-31 00:00:00|\n",
            "|         US|   52738710|R15DUT1VIJ9RJZ|B0053BQN34|     577628462|uDraw Gametablet ...|     Video Games|          2|            0|          0|   N|                Y|We have tried it ...|This was way too ...|2015-08-31 00:00:00|\n",
            "|         US|   10556786|R3IMF2MQ3OU9ZM|B002I0HIMI|     988218515|NBA 2K12(Covers M...|     Video Games|          4|            0|          0|   N|                Y|          Four Stars|Works great good ...|2015-08-31 00:00:00|\n",
            "|         US|    2963837|R23H79DHOZTYAU|B0081EH12M|     770100932|New Trigger Grips...|     Video Games|          1|            1|          1|   N|                Y|Now i have to buy...|It did not fit th...|2015-08-31 00:00:00|\n",
            "|         US|   23092109| RIV24EQAIXA4O|B005FMLZQQ|      24647669|Xbox 360 Media Re...|     Video Games|          5|            0|          0|   N|                Y|          Five Stars|perfect lightweig...|2015-08-31 00:00:00|\n",
            "|         US|   23091728|R3UCNGYDVN24YB|B002BSA388|      33706205|Super Mario Galaxy 2|     Video Games|          5|            0|          0|   N|                Y|          Five Stars|               great|2015-08-31 00:00:00|\n",
            "|         US|   10712640| RUL4H4XTTN2DY|B00BUSLSAC|     829667834|Nintendo 3DS XL -...|     Video Games|          5|            0|          0|   N|                Y|          Five Stars|Works beautifully...|2015-08-31 00:00:00|\n",
            "|         US|   17455376|R20JF7Z4DHTNX5|B00KWF38AW|     110680188|Captain Toad:  Tr...|     Video Games|          5|            0|          0|   N|                Y|          Five Stars|Kids loved the ga...|2015-08-31 00:00:00|\n",
            "|         US|   14754850|R2T1AJ5MFI2260|B00BRQJYA8|     616463426|Lego Batman 2: DC...|     Video Games|          4|            0|          0|   N|                Y|          Four Stars|           Goodngame|2015-08-31 00:00:00|\n",
            "+-----------+-----------+--------------+----------+--------------+--------------------+----------------+-----------+-------------+-----------+----+-----------------+--------------------+--------------------+-------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "### Create DataFrames to match tables"
      ],
      "metadata": {
        "id": "Dp8qduEN5Jck"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.sql.functions import to_date\n",
        "# Read in the Review dataset as a DataFrame\n"
      ],
      "metadata": {
        "id": "Lbei6Aplzk9v"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Create the customers_table DataFrame\n",
        "customers_df = df.groupby(\"customer_id\").agg({\"customer_id\": \"count\"}).withColumnRenamed(\"count(customer_id)\", \"customer_count\")"
      ],
      "metadata": {
        "id": "9hvGpJZ4zoSR"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "customers_df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "ymocr-Jk-1CM",
        "outputId": "cee6e546-01cf-4ee2-c7e6-52c71fd649fc"
      },
      "execution_count": 18,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-----------+--------------+\n",
            "|customer_id|customer_count|\n",
            "+-----------+--------------+\n",
            "|   48670265|             1|\n",
            "|   49103216|             2|\n",
            "|    1131200|             1|\n",
            "|   43076447|             2|\n",
            "|   46261368|             1|\n",
            "|    4883305|             5|\n",
            "|   41192649|             1|\n",
            "|   40985731|             7|\n",
            "|   10437900|             2|\n",
            "|   22245671|             1|\n",
            "|    2574873|             1|\n",
            "|    4696154|             1|\n",
            "|    5621202|             1|\n",
            "|    5871933|             2|\n",
            "|   44089812|             1|\n",
            "|    2845910|             1|\n",
            "|    5274369|             1|\n",
            "|   39069693|             2|\n",
            "|     137793|             1|\n",
            "|   31914942|             3|\n",
            "+-----------+--------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Create the products_table DataFrame and drop duplicates. \n",
        "products_df = df.select([\"product_id\", \"product_title\"]).drop_duplicates([\"product_id\"])"
      ],
      "metadata": {
        "id": "TWUSsK3rzrGH"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "products_df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "8tGd1lSy-6ok",
        "outputId": "8e493005-a69c-401b-c6ca-3cdff0cb3001"
      },
      "execution_count": 19,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----------+--------------------+\n",
            "|product_id|       product_title|\n",
            "+----------+--------------------+\n",
            "|006056038X|        Bitter Truth|\n",
            "|007876355X|Skylanders Giants...|\n",
            "|0078764343|Medal of Honor: W...|\n",
            "|0133617300|Disney Infinity F...|\n",
            "|0182116255|2-Pack Black Plas...|\n",
            "|0226739368|Into the Cool: En...|\n",
            "|0324411812|  Xbox 360 MAS STICK|\n",
            "|0439394422|school house rock...|\n",
            "|0439573947|Need for Speed Ra...|\n",
            "|0439574447|Thomas & Friends ...|\n",
            "|0439591511|Zapper- One Wicke...|\n",
            "|0439591538|Nancy Drew 3D Int...|\n",
            "|0439674492|Mario Paint & Sup...|\n",
            "|0439715571|     Mortal Kombat 4|\n",
            "|0439862108|Narnia the Lion, ...|\n",
            "|0439905419|Sid Meier's Pirat...|\n",
            "|0538966610|Activision Skylan...|\n",
            "|0609808095|The Book of Elean...|\n",
            "|0662476565|Sony Playstation ...|\n",
            "|0700026657|           Anno 2070|\n",
            "+----------+--------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "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\")])"
      ],
      "metadata": {
        "id": "mWNhcuiXzrtG"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "review_id_df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "OX_joEY9_J0x",
        "outputId": "cbd2f03a-129f-4002-988d-28ed7e02be49"
      },
      "execution_count": 20,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "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",
            "|R3AQQ4YUKJWBA6|    2377552|B002UBI6W6|     328764615| 2015-08-31|\n",
            "|R2F0POU5K6F73F|   17521011|B008XHCLFO|      24234603| 2015-08-31|\n",
            "|R3VNR804HYSMR6|   19676307|B00BRA9R6A|     682267517| 2015-08-31|\n",
            "| R3GZTM72WA2QH|     224068|B009EPWJLA|     435241890| 2015-08-31|\n",
            "| RNQOY62705W1K|   48467989|B0000AV7GB|     256572651| 2015-08-31|\n",
            "|R1VTIA3JTYBY02|     106569|B00008KTNN|     384411423| 2015-08-31|\n",
            "|R29DOU8791QZL8|   48269642|B000A3IA0Y|     472622859| 2015-08-31|\n",
            "|R15DUT1VIJ9RJZ|   52738710|B0053BQN34|     577628462| 2015-08-31|\n",
            "|R3IMF2MQ3OU9ZM|   10556786|B002I0HIMI|     988218515| 2015-08-31|\n",
            "|R23H79DHOZTYAU|    2963837|B0081EH12M|     770100932| 2015-08-31|\n",
            "| RIV24EQAIXA4O|   23092109|B005FMLZQQ|      24647669| 2015-08-31|\n",
            "|R3UCNGYDVN24YB|   23091728|B002BSA388|      33706205| 2015-08-31|\n",
            "| RUL4H4XTTN2DY|   10712640|B00BUSLSAC|     829667834| 2015-08-31|\n",
            "|R20JF7Z4DHTNX5|   17455376|B00KWF38AW|     110680188| 2015-08-31|\n",
            "|R2T1AJ5MFI2260|   14754850|B00BRQJYA8|     616463426| 2015-08-31|\n",
            "+--------------+-----------+----------+--------------+-----------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Create the vine_table. DataFrame\n",
        "vine_df = df.select([\"review_id\", \"star_rating\", \"helpful_votes\", \"total_votes\", \"vine\", \"verified_purchase\"])"
      ],
      "metadata": {
        "id": "qx3VbNiDzwXY"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "vine_df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "ciKBJ2TE_QgB",
        "outputId": "07323e80-ce2d-43a0-ad75-8d710c2c2f8f"
      },
      "execution_count": 21,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "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",
            "|R3AQQ4YUKJWBA6|          1|            0|          0|   N|                Y|\n",
            "|R2F0POU5K6F73F|          5|            0|          0|   N|                Y|\n",
            "|R3VNR804HYSMR6|          5|            0|          0|   N|                Y|\n",
            "| R3GZTM72WA2QH|          5|            0|          0|   N|                Y|\n",
            "| RNQOY62705W1K|          4|            0|          0|   N|                Y|\n",
            "|R1VTIA3JTYBY02|          5|            0|          0|   N|                N|\n",
            "|R29DOU8791QZL8|          1|            0|          0|   N|                Y|\n",
            "|R15DUT1VIJ9RJZ|          2|            0|          0|   N|                Y|\n",
            "|R3IMF2MQ3OU9ZM|          4|            0|          0|   N|                Y|\n",
            "|R23H79DHOZTYAU|          1|            1|          1|   N|                Y|\n",
            "| RIV24EQAIXA4O|          5|            0|          0|   N|                Y|\n",
            "|R3UCNGYDVN24YB|          5|            0|          0|   N|                Y|\n",
            "| RUL4H4XTTN2DY|          5|            0|          0|   N|                Y|\n",
            "|R20JF7Z4DHTNX5|          5|            0|          0|   N|                Y|\n",
            "|R2T1AJ5MFI2260|          4|            0|          0|   N|                Y|\n",
            "+--------------+-----------+-------------+-----------+----+-----------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "### Connect to the AWS RDS instance and write each DataFrame to its table"
      ],
      "metadata": {
        "id": "DOGxRoJm5Rv-"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# Configure settings for RDS\n",
        "mode = \"append\"\n",
        "jdbc_url=\"jdbc:postgresql://dataviz.cniovpx5xoeo.us-east-2.rds.amazonaws.com:5432/amazon_etl\"\n",
        "config = {\"user\":\"postgres\", \n",
        "          \"password\": \"\", \n",
        "          \"driver\":\"org.postgresql.Driver\"}"
      ],
      "metadata": {
        "id": "Nnz_RncAzx77"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "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)"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 1000
        },
        "id": "G-y17iGbz08u",
        "outputId": "3295f73f-939d-4a2d-bb72-01000bae22c0"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "error",
          "ename": "Py4JJavaError",
          "evalue": "ignored",
          "traceback": [
            "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
            "\u001b[0;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
            "\u001b[0;32m<ipython-input-17-777dd587f0a7>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0;31m# Write review_id_df to table in RDS\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 2\u001b[0;31m \u001b[0mreview_id_df\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mwrite\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjdbc\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0murl\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mjdbc_url\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mtable\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m'review_id_table'\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mmode\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mmode\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mproperties\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mconfig\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
            "\u001b[0;32m/content/spark-3.3.0-bin-hadoop3/python/pyspark/sql/readwriter.py\u001b[0m in \u001b[0;36mjdbc\u001b[0;34m(self, url, table, mode, properties)\u001b[0m\n\u001b[1;32m   1338\u001b[0m         \u001b[0;32mfor\u001b[0m \u001b[0mk\u001b[0m \u001b[0;32min\u001b[0m \u001b[0mproperties\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1339\u001b[0m             \u001b[0mjprop\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msetProperty\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mk\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mproperties\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0mk\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1340\u001b[0;31m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mmode\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mmode\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jwrite\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjdbc\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0murl\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mtable\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mjprop\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1341\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1342\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.3.0-bin-hadoop3/python/lib/py4j-0.10.9.5-src.zip/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1320\u001b[0m         \u001b[0manswer\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1321\u001b[0m         return_value = get_return_value(\n\u001b[0;32m-> 1322\u001b[0;31m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[1;32m   1323\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1324\u001b[0m         \u001b[0;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[0;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.3.0-bin-hadoop3/python/pyspark/sql/utils.py\u001b[0m in \u001b[0;36mdeco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m    188\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mdeco\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m*\u001b[0m\u001b[0ma\u001b[0m\u001b[0;34m:\u001b[0m \u001b[0mAny\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[0;34m:\u001b[0m \u001b[0mAny\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;34m->\u001b[0m \u001b[0mAny\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    189\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 190\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mf\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m*\u001b[0m\u001b[0ma\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    191\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mPy4JJavaError\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    192\u001b[0m             \u001b[0mconverted\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mconvert_exception\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0me\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjava_exception\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.3.0-bin-hadoop3/python/lib/py4j-0.10.9.5-src.zip/py4j/protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[0;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[1;32m    326\u001b[0m                 raise Py4JJavaError(\n\u001b[1;32m    327\u001b[0m                     \u001b[0;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 328\u001b[0;31m                     format(target_id, \".\", name), value)\n\u001b[0m\u001b[1;32m    329\u001b[0m             \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    330\u001b[0m                 raise Py4JError(\n",
            "\u001b[0;31mPy4JJavaError\u001b[0m: An error occurred while calling o100.jdbc.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 13.0 failed 1 times, most recent failure: Lost task 0.0 in stage 13.0 (TID 13) (47bee1d3f766 executor driver): java.sql.BatchUpdateException: Batch entry 0 INSERT INTO review_id_table (\"review_id\",\"customer_id\",\"product_id\",\"product_parent\",\"review_date\") VALUES ('RTIS3L2M1F5SM',12039526,'B001CXYMFS',737716809,'2015-08-31 +00'::date) was aborted: ERROR: duplicate key value violates unique constraint \"review_id_table_pkey\"\n  Detail: Key (review_id)=(RTIS3L2M1F5SM) already exists.  Call getNextException to see other errors in the batch.\n\tat org.postgresql.jdbc.BatchResultHandler.handleError(BatchResultHandler.java:169)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2286)\n\tat org.postgresql.core.v3.QueryExecutorImpl.flushIfDeadlockRisk(QueryExecutorImpl.java:1404)\n\tat org.postgresql.core.v3.QueryExecutorImpl.sendQuery(QueryExecutorImpl.java:1429)\n\tat org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:507)\n\tat org.postgresql.jdbc.PgStatement.internalExecuteBatch(PgStatement.java:870)\n\tat org.postgresql.jdbc.PgStatement.executeBatch(PgStatement.java:893)\n\tat org.postgresql.jdbc.PgPreparedStatement.executeBatch(PgPreparedStatement.java:1639)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.savePartition(JdbcUtils.scala:708)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$saveTable$1(JdbcUtils.scala:868)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$saveTable$1$adapted(JdbcUtils.scala:867)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2(RDD.scala:1011)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2$adapted(RDD.scala:1011)\n\tat org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2268)\n\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:136)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\nCaused by: org.postgresql.util.PSQLException: ERROR: duplicate key value violates unique constraint \"review_id_table_pkey\"\n  Detail: Key (review_id)=(RTIS3L2M1F5SM) already exists.\n\tat org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2553)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2285)\n\t... 20 more\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2672)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2608)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2607)\n\tat scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)\n\tat scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)\n\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)\n\tat org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2607)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:1182)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:1182)\n\tat scala.Option.foreach(Option.scala:407)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1182)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2860)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2802)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2791)\n\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)\n\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:952)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2228)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2249)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2268)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2293)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$1(RDD.scala:1011)\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)\n\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:406)\n\tat org.apache.spark.rdd.RDD.foreachPartition(RDD.scala:1009)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.saveTable(JdbcUtils.scala:867)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:70)\n\tat org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:45)\n\tat org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:75)\n\tat org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:73)\n\tat org.apache.spark.sql.execution.command.ExecutedCommandExec.executeCollect(commands.scala:84)\n\tat org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.$anonfun$applyOrElse$1(QueryExecution.scala:98)\n\tat org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:109)\n\tat org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:169)\n\tat org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:95)\n\tat org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:779)\n\tat org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)\n\tat org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:98)\n\tat org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:94)\n\tat org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:584)\n\tat org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:176)\n\tat org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:584)\n\tat org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:30)\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)\n\tat org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)\n\tat org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)\n\tat org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)\n\tat org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:560)\n\tat org.apache.spark.sql.execution.QueryExecution.eagerlyExecuteCommands(QueryExecution.scala:94)\n\tat org.apache.spark.sql.execution.QueryExecution.commandExecuted$lzycompute(QueryExecution.scala:81)\n\tat org.apache.spark.sql.execution.QueryExecution.commandExecuted(QueryExecution.scala:79)\n\tat org.apache.spark.sql.execution.QueryExecution.assertCommandExecuted(QueryExecution.scala:116)\n\tat org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:860)\n\tat org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:390)\n\tat org.apache.spark.sql.DataFrameWriter.saveInternal(DataFrameWriter.scala:363)\n\tat org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:247)\n\tat org.apache.spark.sql.DataFrameWriter.jdbc(DataFrameWriter.scala:757)\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\n\tat java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.base/java.lang.reflect.Method.invoke(Method.java:566)\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\n\tat py4j.Gateway.invoke(Gateway.java:282)\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n\tat py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)\n\tat py4j.ClientServerConnection.run(ClientServerConnection.java:106)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\nCaused by: java.sql.BatchUpdateException: Batch entry 0 INSERT INTO review_id_table (\"review_id\",\"customer_id\",\"product_id\",\"product_parent\",\"review_date\") VALUES ('RTIS3L2M1F5SM',12039526,'B001CXYMFS',737716809,'2015-08-31 +00'::date) was aborted: ERROR: duplicate key value violates unique constraint \"review_id_table_pkey\"\n  Detail: Key (review_id)=(RTIS3L2M1F5SM) already exists.  Call getNextException to see other errors in the batch.\n\tat org.postgresql.jdbc.BatchResultHandler.handleError(BatchResultHandler.java:169)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2286)\n\tat org.postgresql.core.v3.QueryExecutorImpl.flushIfDeadlockRisk(QueryExecutorImpl.java:1404)\n\tat org.postgresql.core.v3.QueryExecutorImpl.sendQuery(QueryExecutorImpl.java:1429)\n\tat org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:507)\n\tat org.postgresql.jdbc.PgStatement.internalExecuteBatch(PgStatement.java:870)\n\tat org.postgresql.jdbc.PgStatement.executeBatch(PgStatement.java:893)\n\tat org.postgresql.jdbc.PgPreparedStatement.executeBatch(PgPreparedStatement.java:1639)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.savePartition(JdbcUtils.scala:708)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$saveTable$1(JdbcUtils.scala:868)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$saveTable$1$adapted(JdbcUtils.scala:867)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2(RDD.scala:1011)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2$adapted(RDD.scala:1011)\n\tat org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2268)\n\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:136)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\t... 1 more\nCaused by: org.postgresql.util.PSQLException: ERROR: duplicate key value violates unique constraint \"review_id_table_pkey\"\n  Detail: Key (review_id)=(RTIS3L2M1F5SM) already exists.\n\tat org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2553)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2285)\n\t... 20 more\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "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)"
      ],
      "metadata": {
        "id": "JEbuQAa92jvG"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "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)"
      ],
      "metadata": {
        "id": "B69Mj5NRz3X2"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "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)"
      ],
      "metadata": {
        "id": "nmVX3g_83gf9"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}