Amazon_Reviews_ETL.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Amazon_Reviews_ETL.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "authorship_tag": "ABX9TyNKtnqJEBmR36+o0Ay/fxH6",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/nhafer88/Amazon_Vine_Analysis/blob/main/Amazon_Reviews_ETL.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "eEXDgDeFMxHg",
        "outputId": "0f0a3078-1899-4a05-ae19-8b62ee4d7191"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "\r0% [Working]\r            \rIgn:1 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  InRelease\n",
            "\r0% [Connecting to archive.ubuntu.com (91.189.88.142)] [Connecting to security.u\r                                                                               \rIgn:2 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  InRelease\n",
            "\r0% [Connecting to archive.ubuntu.com (91.189.88.142)] [Waiting for headers] [Co\r                                                                               \rGet:3 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  Release [696 B]\n",
            "Hit:4 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  Release\n",
            "Get:5 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  Release.gpg [836 B]\n",
            "Get:6 https://cloud.r-project.org/bin/linux/ubuntu bionic-cran40/ InRelease [3,626 B]\n",
            "Get:7 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic InRelease [15.9 kB]\n",
            "Get:8 http://security.ubuntu.com/ubuntu bionic-security InRelease [88.7 kB]\n",
            "Hit:9 http://archive.ubuntu.com/ubuntu bionic InRelease\n",
            "Get:11 http://archive.ubuntu.com/ubuntu bionic-updates InRelease [88.7 kB]\n",
            "Hit:12 http://ppa.launchpad.net/cran/libgit2/ubuntu bionic InRelease\n",
            "Get:13 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  Packages [867 kB]\n",
            "Get:14 http://archive.ubuntu.com/ubuntu bionic-backports InRelease [74.6 kB]\n",
            "Get:15 http://ppa.launchpad.net/deadsnakes/ppa/ubuntu bionic InRelease [15.9 kB]\n",
            "Hit:16 http://ppa.launchpad.net/graphics-drivers/ppa/ubuntu bionic InRelease\n",
            "Get:17 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic/main Sources [1,822 kB]\n",
            "Get:18 http://security.ubuntu.com/ubuntu bionic-security/restricted amd64 Packages [725 kB]\n",
            "Get:19 http://security.ubuntu.com/ubuntu bionic-security/main amd64 Packages [2,498 kB]\n",
            "Get:20 http://archive.ubuntu.com/ubuntu bionic-updates/main amd64 Packages [2,935 kB]\n",
            "Get:21 http://security.ubuntu.com/ubuntu bionic-security/universe amd64 Packages [1,459 kB]\n",
            "Get:22 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic/main amd64 Packages [934 kB]\n",
            "Get:23 http://archive.ubuntu.com/ubuntu bionic-updates/restricted amd64 Packages [758 kB]\n",
            "Get:24 http://archive.ubuntu.com/ubuntu bionic-updates/universe amd64 Packages [2,238 kB]\n",
            "Get:25 http://ppa.launchpad.net/deadsnakes/ppa/ubuntu bionic/main amd64 Packages [46.6 kB]\n",
            "Fetched 14.6 MB in 5s (2,836 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.0.3'\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-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",
      "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": "OF4Ty997NlV9",
        "outputId": "2361cca9-389f-457d-faed-e73b8d7ff651"
      },
      "execution_count": 2,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "--2022-01-15 19:28:40--  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.90MB/s    in 0.2s    \n",
            "\n",
            "2022-01-15 19:28:41 (4.90 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": "yoK2LcW9NyLF"
      },
      "execution_count": 3,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Load Amazon Data in Spark DataFrame\n",
        "from pyspark import SparkFiles\n",
        "url = \"https://s3.amazonaws.com/amazon-reviews-pds/tsv/amazon_reviews_us_Outdoors_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()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "F6hxagwYN7jV",
        "outputId": "aafbd2c8-34b1-4b3a-e9fa-59569868e0f6"
      },
      "execution_count": 4,
      "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|   18446823|R35T75OLUGHL5C|B000NV6H94|     110804376|Stearns Youth Boa...|        Outdoors|          4|            0|          0|   N|                Y|          Four Stars|          GOOD VALUE| 2015-08-31|\n",
            "|         US|   13724367|R2BV735O46BN33|B000IN0W3Y|     624096774|Primal Wear Men's...|        Outdoors|          5|            0|          0|   N|                Y|          Five Stars|  Excellent quality.| 2015-08-31|\n",
            "|         US|   51001958|R2NBEUGPQQGXP1|B008RBJXFM|     278970944|Osprey Hydraulics...|        Outdoors|          4|            0|          0|   N|                Y|Only Flaw Is The Cap|3rd season using ...| 2015-08-31|\n",
            "|         US|   32866903|R17LLAOJ8ITK0S|B00FK8WUQY|     312877650|CamelBak eddy .75...|        Outdoors|          3|            1|          1|   N|                Y|Poor design leads...|poor construction...| 2015-08-31|\n",
            "|         US|   30907790|R39PEQBT5ISEF4|B00EZA3VW0|     305567912|Children Black Re...|        Outdoors|          1|            0|          0|   N|                Y|Very bad quality,...|Very bad quality,...| 2015-08-31|\n",
            "|         US|   20232229|R3GNM3SU9VHJFT|B006JA8WEG|     842306035|Ibera Bicycle Tri...|        Outdoors|          4|            1|          1|   N|                Y|Nice bag. Should ...|Nice bag. Should ...| 2015-08-31|\n",
            "|         US|   17698862| R2Y81OP0EK467|B002PWFSEO|     451480122|Therm-a-Rest Comp...|        Outdoors|          5|            0|          0|   N|                Y|Very comfortable ...|Gave this to my s...| 2015-08-31|\n",
            "|         US|   38486114|R2LFGSI6HAYH5F|B002DZGKHW|     124386306|Sawyer Products P...|        Outdoors|          5|            1|          1|   N|                Y| Worked like a charm|Went on vacation ...| 2015-08-31|\n",
            "|         US|   26319572|R297G6ED1IQO7W|B00ABA08F6|     991442421|Zippo Hand Warmer...|        Outdoors|          5|            1|          1|   N|                Y|Great item. Gets ...|Great item. Gets ...| 2015-08-31|\n",
            "|         US|   27152337| RE27RFC6101N6|B003Z8WIHC|     886483892|Camp Chef Dutch O...|        Outdoors|          5|            0|          0|   N|                Y|Great value for t...|I am so glad I bo...| 2015-08-31|\n",
            "|         US|   12516845|R3BPDME6E94W8Z|B007CP6UK0|     150224054|3CERA Portable Wi...|        Outdoors|          5|            0|          0|   N|                Y|          Five Stars|        good to have| 2015-08-31|\n",
            "|         US|    3225242|R2P08O1RILUOX3|B003V3U9JK|     343847969|Texsport King Kot...|        Outdoors|          3|            0|          0|   N|                Y|Cot set up inconv...|VERY difficult to...| 2015-08-31|\n",
            "|         US|     961839|R37CVAB03PTDVI|B00Y846HN8|     858088629|Wallygadgets 2 Wh...|        Outdoors|          5|            0|          1|   N|                Y|          Five Stars|Thanks excellent ...| 2015-08-31|\n",
            "|         US|   47796452| RAWNWOGXPCPMD|B00IYQ84VY|     474493517|RainStoppers 34-I...|        Outdoors|          5|            0|          0|   N|                Y|          Five Stars|This umbrella is ...| 2015-08-31|\n",
            "|         US|   32004835| R5DYGP6ASX77M|B002MYCKLY|     920014456|Alpha Deluxe Port...|        Outdoors|          5|            0|          0|   N|                Y|          Five Stars|Love it !! I even...| 2015-08-31|\n",
            "|         US|   23972939|R1O0SAOOGF2KG7|B00EZV69JG|     128489321|Speedfil Z4 BTA B...|        Outdoors|          4|            0|          0|   N|                Y|        Good enough.|This is a fine mo...| 2015-08-31|\n",
            "|         US|   40889047|R35NJUT0U3MU3V|B00AWOT3T8|     571303876|O'Brien Kids Plat...|        Outdoors|          5|            0|          0|   N|                Y| Got Up on First Try|We just bought th...| 2015-08-31|\n",
            "|         US|   11244387|R242C08MF9D1AH|B0000AXTID|     739769424|Kwik-Tek F-5R Pla...|        Outdoors|          5|            0|          0|   N|                Y|They go over an a...|I have these on m...| 2015-08-31|\n",
            "|         US|   20121211| R3RYG8TJTO4E2|B00IFHFJXI|     984009972|Ivation Portable ...|        Outdoors|          5|            0|          0|   N|                Y|Greatest Item I b...|This is the best ...| 2015-08-31|\n",
            "|         US|   25657249|R3IKH1DNY0CP9F|B00KFILTWU|     405521681|GreenInsync Repla...|        Outdoors|          2|            0|          0|   N|                Y|I received this p...|I received this p...| 2015-08-31|\n",
            "+-----------+-----------+--------------+----------+--------------+--------------------+----------------+-----------+-------------+-----------+----+-----------------+--------------------+--------------------+-----------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.sql.functions import to_date\n",
        "# Read in the Review dataset as a DataFrame\n",
        "review_df = df.select([\"review_id\",\"customer_id\",\"product_id\",\"product_parent\",\"review_date\"])\n",
        "review_df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "L_rImqNtO12y",
        "outputId": "85e4db77-616a-403a-c0a1-1358f68d1395"
      },
      "execution_count": 7,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------------+-----------+----------+--------------+-----------+\n",
            "|     review_id|customer_id|product_id|product_parent|review_date|\n",
            "+--------------+-----------+----------+--------------+-----------+\n",
            "|R35T75OLUGHL5C|   18446823|B000NV6H94|     110804376| 2015-08-31|\n",
            "|R2BV735O46BN33|   13724367|B000IN0W3Y|     624096774| 2015-08-31|\n",
            "|R2NBEUGPQQGXP1|   51001958|B008RBJXFM|     278970944| 2015-08-31|\n",
            "|R17LLAOJ8ITK0S|   32866903|B00FK8WUQY|     312877650| 2015-08-31|\n",
            "|R39PEQBT5ISEF4|   30907790|B00EZA3VW0|     305567912| 2015-08-31|\n",
            "|R3GNM3SU9VHJFT|   20232229|B006JA8WEG|     842306035| 2015-08-31|\n",
            "| R2Y81OP0EK467|   17698862|B002PWFSEO|     451480122| 2015-08-31|\n",
            "|R2LFGSI6HAYH5F|   38486114|B002DZGKHW|     124386306| 2015-08-31|\n",
            "|R297G6ED1IQO7W|   26319572|B00ABA08F6|     991442421| 2015-08-31|\n",
            "| RE27RFC6101N6|   27152337|B003Z8WIHC|     886483892| 2015-08-31|\n",
            "|R3BPDME6E94W8Z|   12516845|B007CP6UK0|     150224054| 2015-08-31|\n",
            "|R2P08O1RILUOX3|    3225242|B003V3U9JK|     343847969| 2015-08-31|\n",
            "|R37CVAB03PTDVI|     961839|B00Y846HN8|     858088629| 2015-08-31|\n",
            "| RAWNWOGXPCPMD|   47796452|B00IYQ84VY|     474493517| 2015-08-31|\n",
            "| R5DYGP6ASX77M|   32004835|B002MYCKLY|     920014456| 2015-08-31|\n",
            "|R1O0SAOOGF2KG7|   23972939|B00EZV69JG|     128489321| 2015-08-31|\n",
            "|R35NJUT0U3MU3V|   40889047|B00AWOT3T8|     571303876| 2015-08-31|\n",
            "|R242C08MF9D1AH|   11244387|B0000AXTID|     739769424| 2015-08-31|\n",
            "| R3RYG8TJTO4E2|   20121211|B00IFHFJXI|     984009972| 2015-08-31|\n",
            "|R3IKH1DNY0CP9F|   25657249|B00KFILTWU|     405521681| 2015-08-31|\n",
            "+--------------+-----------+----------+--------------+-----------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "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\")\n",
        "customers_df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "aN56IA29PGNm",
        "outputId": "7daf85c2-2e3f-44b3-8ddd-94d57e332b61"
      },
      "execution_count": 10,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-----------+--------------+\n",
            "|customer_id|customer_count|\n",
            "+-----------+--------------+\n",
            "|   43679767|             1|\n",
            "|   32024654|             1|\n",
            "|   52913169|             1|\n",
            "|   24297214|             1|\n",
            "|   26096454|             1|\n",
            "|   38247118|             1|\n",
            "|   44009906|             1|\n",
            "|    1753876|             1|\n",
            "|   26195644|             1|\n",
            "|   23042837|             1|\n",
            "|   11487525|             2|\n",
            "|   28258386|             1|\n",
            "|   42560427|             2|\n",
            "|   24540309|             1|\n",
            "|    1967239|             1|\n",
            "|   14217455|             1|\n",
            "|   20289484|            26|\n",
            "|   47103434|             1|\n",
            "|   28638887|             1|\n",
            "|     740678|             1|\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()\n",
        "products_df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "6hIMD7c-TzLL",
        "outputId": "41bf89d2-5b7f-4be0-9c40-89f12f2af4fa"
      },
      "execution_count": 13,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----------+--------------------+\n",
            "|product_id|       product_title|\n",
            "+----------+--------------------+\n",
            "|B00IFHFJXI|Ivation Portable ...|\n",
            "|B00WG0J0D0|JanSport Superbre...|\n",
            "|B00V15AUN0|Nickelodeon Paw P...|\n",
            "|B00FUWSTI8|Bago Lightweight ...|\n",
            "|B003FV94NA|Michelin Lithion ...|\n",
            "|B00WIK04HO|Ultra Bright Camp...|\n",
            "|B00J2HSCM0|High Sierra Tank ...|\n",
            "|B009I6NSR4|Black Veil Brides...|\n",
            "|B001GSHSLE|Stansport 191 App...|\n",
            "|B00L2IO9M4|Columbia Sportswe...|\n",
            "|B00KY7IM7W|Nalgene 32 Oz Wid...|\n",
            "|B00TV5JCTK|Rollerblade ABEC ...|\n",
            "|B00B9D071Y|BUFF UV Multifunc...|\n",
            "|B00F9IGIKO|Condor Tactical F...|\n",
            "|B004X55L9I|Hydro Flask Insul...|\n",
            "|B00LORROIY|Scuba Choice Divi...|\n",
            "|B00AATRU8G|Kelty Redwing 44 ...|\n",
            "|B00HMCYWEO|Dakine Explorer L...|\n",
            "|B004DK1CM8|Hot Headz 12V Hea...|\n",
            "|B00T4W6SSS|Fits Sock Light H...|\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\")])\n",
        "review_id_df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "ZuT_D9YrVEd-",
        "outputId": "9de7c4ba-935c-47c0-830d-cf27a35bbb14"
      },
      "execution_count": 18,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------------+-----------+----------+--------------+-----------+\n",
            "|     review_id|customer_id|product_id|product_parent|review_date|\n",
            "+--------------+-----------+----------+--------------+-----------+\n",
            "|R35T75OLUGHL5C|   18446823|B000NV6H94|     110804376| 2015-08-31|\n",
            "|R2BV735O46BN33|   13724367|B000IN0W3Y|     624096774| 2015-08-31|\n",
            "|R2NBEUGPQQGXP1|   51001958|B008RBJXFM|     278970944| 2015-08-31|\n",
            "|R17LLAOJ8ITK0S|   32866903|B00FK8WUQY|     312877650| 2015-08-31|\n",
            "|R39PEQBT5ISEF4|   30907790|B00EZA3VW0|     305567912| 2015-08-31|\n",
            "|R3GNM3SU9VHJFT|   20232229|B006JA8WEG|     842306035| 2015-08-31|\n",
            "| R2Y81OP0EK467|   17698862|B002PWFSEO|     451480122| 2015-08-31|\n",
            "|R2LFGSI6HAYH5F|   38486114|B002DZGKHW|     124386306| 2015-08-31|\n",
            "|R297G6ED1IQO7W|   26319572|B00ABA08F6|     991442421| 2015-08-31|\n",
            "| RE27RFC6101N6|   27152337|B003Z8WIHC|     886483892| 2015-08-31|\n",
            "|R3BPDME6E94W8Z|   12516845|B007CP6UK0|     150224054| 2015-08-31|\n",
            "|R2P08O1RILUOX3|    3225242|B003V3U9JK|     343847969| 2015-08-31|\n",
            "|R37CVAB03PTDVI|     961839|B00Y846HN8|     858088629| 2015-08-31|\n",
            "| RAWNWOGXPCPMD|   47796452|B00IYQ84VY|     474493517| 2015-08-31|\n",
            "| R5DYGP6ASX77M|   32004835|B002MYCKLY|     920014456| 2015-08-31|\n",
            "|R1O0SAOOGF2KG7|   23972939|B00EZV69JG|     128489321| 2015-08-31|\n",
            "|R35NJUT0U3MU3V|   40889047|B00AWOT3T8|     571303876| 2015-08-31|\n",
            "|R242C08MF9D1AH|   11244387|B0000AXTID|     739769424| 2015-08-31|\n",
            "| R3RYG8TJTO4E2|   20121211|B00IFHFJXI|     984009972| 2015-08-31|\n",
            "|R3IKH1DNY0CP9F|   25657249|B00KFILTWU|     405521681| 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\"])\n",
        "vine_df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "Df8vL1UsXhtR",
        "outputId": "3fda30b9-174d-45c1-ad85-d6756c3d067c"
      },
      "execution_count": 41,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------------+-----------+-------------+-----------+----+-----------------+\n",
            "|     review_id|star_rating|helpful_votes|total_votes|vine|verified_purchase|\n",
            "+--------------+-----------+-------------+-----------+----+-----------------+\n",
            "|R35T75OLUGHL5C|          4|            0|          0|   N|                Y|\n",
            "|R2BV735O46BN33|          5|            0|          0|   N|                Y|\n",
            "|R2NBEUGPQQGXP1|          4|            0|          0|   N|                Y|\n",
            "|R17LLAOJ8ITK0S|          3|            1|          1|   N|                Y|\n",
            "|R39PEQBT5ISEF4|          1|            0|          0|   N|                Y|\n",
            "|R3GNM3SU9VHJFT|          4|            1|          1|   N|                Y|\n",
            "| R2Y81OP0EK467|          5|            0|          0|   N|                Y|\n",
            "|R2LFGSI6HAYH5F|          5|            1|          1|   N|                Y|\n",
            "|R297G6ED1IQO7W|          5|            1|          1|   N|                Y|\n",
            "| RE27RFC6101N6|          5|            0|          0|   N|                Y|\n",
            "|R3BPDME6E94W8Z|          5|            0|          0|   N|                Y|\n",
            "|R2P08O1RILUOX3|          3|            0|          0|   N|                Y|\n",
            "|R37CVAB03PTDVI|          5|            0|          1|   N|                Y|\n",
            "| RAWNWOGXPCPMD|          5|            0|          0|   N|                Y|\n",
            "| R5DYGP6ASX77M|          5|            0|          0|   N|                Y|\n",
            "|R1O0SAOOGF2KG7|          4|            0|          0|   N|                Y|\n",
            "|R35NJUT0U3MU3V|          5|            0|          0|   N|                Y|\n",
            "|R242C08MF9D1AH|          5|            0|          0|   N|                Y|\n",
            "| R3RYG8TJTO4E2|          5|            0|          0|   N|                Y|\n",
            "|R3IKH1DNY0CP9F|          2|            0|          0|   N|                Y|\n",
            "+--------------+-----------+-------------+-----------+----+-----------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Store environmental variable\n",
        "#from getpass import getpass\n",
        "#password = getpass('Enter database password')\n",
        "# Configure settings for RDS\n",
        "mode = \"overwrite\"\n",
        "jdbc_url=\"jdbc:postgresql://sellby-data.cbgckvl0afyb.us-east-2.rds.amazonaws.com:5432/postgres\"\n",
        "config = {\"user\":\"postgres\",\n",
        "          \"password\": \"Yellow1990!\",\n",
        "          \"driver\":\"org.postgresql.Driver\"}"
      ],
      "metadata": {
        "id": "5DLfLTKFdF1D"
      },
      "execution_count": 45,
      "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": "gtyehdWtePti",
        "outputId": "e0c41eca-df3d-40ba-bbad-cad48737069d"
      },
      "execution_count": 37,
      "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-37-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.0.3-bin-hadoop2.7/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   1083\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   1084\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-> 1085\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   1086\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1087\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.0.3-bin-hadoop2.7/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1303\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   1304\u001b[0m         return_value = get_return_value(\n\u001b[0;32m-> 1305\u001b[0;31m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[1;32m   1306\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1307\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.0.3-bin-hadoop2.7/python/pyspark/sql/utils.py\u001b[0m in \u001b[0;36mdeco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m    126\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[0;34m**\u001b[0m\u001b[0mkw\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[1;32m    127\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--> 128\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    129\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mpy4j\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mprotocol\u001b[0m\u001b[0;34m.\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    130\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.0.3-bin-hadoop2.7/python/lib/py4j-0.10.9-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 o182.jdbc.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 14.0 failed 1 times, most recent failure: Lost task 0.0 in stage 14.0 (TID 14, f46c4c42080f, executor driver): java.sql.BatchUpdateException: Batch entry 0 INSERT INTO review_id_table (\"review_id\",\"customer_id\",\"product_id\",\"product_parent\",\"review_date\") VALUES ('R1QVL8T7XOEZHK',20192571,'B003SXLW7Q',255478975,'2015-08-31 +00') was aborted: ERROR: duplicate key value violates unique constraint \"review_id_table_pkey\"\n  Detail: Key (review_id)=(R1QVL8T7XOEZHK) 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:699)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$saveTable$1(JdbcUtils.scala:871)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$saveTable$1$adapted(JdbcUtils.scala:869)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2(RDD.scala:994)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2$adapted(RDD.scala:994)\n\tat org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2154)\n\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:127)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:463)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:466)\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)=(R1QVL8T7XOEZHK) 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:2059)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2008)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2007)\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:2007)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:973)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:973)\n\tat scala.Option.foreach(Option.scala:407)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:973)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2239)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2188)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2177)\n\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)\n\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:775)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2114)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2135)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2154)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2179)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$1(RDD.scala:994)\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:388)\n\tat org.apache.spark.rdd.RDD.foreachPartition(RDD.scala:992)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.saveTable(JdbcUtils.scala:869)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:68)\n\tat org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:46)\n\tat org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)\n\tat org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)\n\tat org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:90)\n\tat org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:180)\n\tat org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:218)\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)\n\tat org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:215)\n\tat org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:176)\n\tat org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:127)\n\tat org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:126)\n\tat org.apache.spark.sql.DataFrameWriter.$anonfun$runCommand$1(DataFrameWriter.scala:962)\n\tat org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:100)\n\tat org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:160)\n\tat org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:87)\n\tat org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:767)\n\tat org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)\n\tat org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:962)\n\tat org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:414)\n\tat org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:398)\n\tat org.apache.spark.sql.DataFrameWriter.jdbc(DataFrameWriter.scala:790)\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.GatewayConnection.run(GatewayConnection.java:238)\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 ('R1QVL8T7XOEZHK',20192571,'B003SXLW7Q',255478975,'2015-08-31 +00') was aborted: ERROR: duplicate key value violates unique constraint \"review_id_table_pkey\"\n  Detail: Key (review_id)=(R1QVL8T7XOEZHK) 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:699)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$saveTable$1(JdbcUtils.scala:871)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$saveTable$1$adapted(JdbcUtils.scala:869)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2(RDD.scala:994)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2$adapted(RDD.scala:994)\n\tat org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2154)\n\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:127)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:463)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:466)\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)=(R1QVL8T7XOEZHK) 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": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 1000
        },
        "id": "ecuin9zTqTkz",
        "outputId": "a7788880-6d13-491c-a084-433a6d28b33d"
      },
      "execution_count": 38,
      "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-38-f9bb330a0bb2>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0;31m# Write products_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[1;32m      2\u001b[0m \u001b[0;31m# about 3 min\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 3\u001b[0;31m \u001b[0mproducts_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'products_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.0.3-bin-hadoop2.7/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   1083\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   1084\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-> 1085\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   1086\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1087\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.0.3-bin-hadoop2.7/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1303\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   1304\u001b[0m         return_value = get_return_value(\n\u001b[0;32m-> 1305\u001b[0;31m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[1;32m   1306\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1307\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.0.3-bin-hadoop2.7/python/pyspark/sql/utils.py\u001b[0m in \u001b[0;36mdeco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m    126\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[0;34m**\u001b[0m\u001b[0mkw\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[1;32m    127\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--> 128\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    129\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mpy4j\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mprotocol\u001b[0m\u001b[0;34m.\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    130\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.0.3-bin-hadoop2.7/python/lib/py4j-0.10.9-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 o188.jdbc.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 22 in stage 16.0 failed 1 times, most recent failure: Lost task 22.0 in stage 16.0 (TID 38, f46c4c42080f, executor driver): java.sql.BatchUpdateException: Batch entry 905 INSERT INTO products_table (\"product_id\",\"product_title\") VALUES ('B00M9NO3HM','Bomber Irie Bomb Floating Sunglasses\tOutdoors\t4\t0\t0\tN\tY\tFavorite Christmas gift\tBought these for my 14yr old son who paddles outrigger and he loves them. Comfortable and they float if they fall off.\t2015-01-04') was aborted: ERROR: duplicate key value violates unique constraint \"products_table_pkey\"\n  Detail: Key (product_id)=(B00M9NO3HM) 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.execute(QueryExecutorImpl.java:521)\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:699)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$saveTable$1(JdbcUtils.scala:871)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$saveTable$1$adapted(JdbcUtils.scala:869)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2(RDD.scala:994)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2$adapted(RDD.scala:994)\n\tat org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2154)\n\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:127)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:463)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:466)\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 \"products_table_pkey\"\n  Detail: Key (product_id)=(B00M9NO3HM) 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... 18 more\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2059)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2008)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2007)\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:2007)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:973)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:973)\n\tat scala.Option.foreach(Option.scala:407)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:973)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2239)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2188)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2177)\n\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)\n\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:775)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2114)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2135)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2154)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2179)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$1(RDD.scala:994)\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:388)\n\tat org.apache.spark.rdd.RDD.foreachPartition(RDD.scala:992)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.saveTable(JdbcUtils.scala:869)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:68)\n\tat org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:46)\n\tat org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)\n\tat org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)\n\tat org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:90)\n\tat org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:180)\n\tat org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:218)\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)\n\tat org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:215)\n\tat org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:176)\n\tat org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:127)\n\tat org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:126)\n\tat org.apache.spark.sql.DataFrameWriter.$anonfun$runCommand$1(DataFrameWriter.scala:962)\n\tat org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:100)\n\tat org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:160)\n\tat org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:87)\n\tat org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:767)\n\tat org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)\n\tat org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:962)\n\tat org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:414)\n\tat org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:398)\n\tat org.apache.spark.sql.DataFrameWriter.jdbc(DataFrameWriter.scala:790)\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.GatewayConnection.run(GatewayConnection.java:238)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\nCaused by: java.sql.BatchUpdateException: Batch entry 905 INSERT INTO products_table (\"product_id\",\"product_title\") VALUES ('B00M9NO3HM','Bomber Irie Bomb Floating Sunglasses\tOutdoors\t4\t0\t0\tN\tY\tFavorite Christmas gift\tBought these for my 14yr old son who paddles outrigger and he loves them. Comfortable and they float if they fall off.\t2015-01-04') was aborted: ERROR: duplicate key value violates unique constraint \"products_table_pkey\"\n  Detail: Key (product_id)=(B00M9NO3HM) 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.execute(QueryExecutorImpl.java:521)\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:699)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$saveTable$1(JdbcUtils.scala:871)\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$saveTable$1$adapted(JdbcUtils.scala:869)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2(RDD.scala:994)\n\tat org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2$adapted(RDD.scala:994)\n\tat org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2154)\n\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:127)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:463)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:466)\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 \"products_table_pkey\"\n  Detail: Key (product_id)=(B00M9NO3HM) 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... 18 more\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "customers_df.write.jdbc(url=jdbc_url, table='customers_table', mode=mode, properties=config)"
      ],
      "metadata": {
        "id": "bf42NTrqrMu_"
      },
      "execution_count": 39,
      "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": "5mGn8GoOsksF"
      },
      "execution_count": 46,
      "outputs": []
    }
  ]
}