Amazon_Reviews_ETL.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "collapsed_sections": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "nteract": {
      "version": "0.12.3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "code",
      "metadata": {
        "id": "V58rxea0HqSa",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "8e7b7374-f6f9-4a1b-ba41-153ca1f0f8b9"
      },
      "source": [
        "import os\n",
        "# Find the latest version of spark 3.0  from http://www.apache.org/dist/spark/ \n",
        "# and enter as the spark version\n",
        "# For example:\n",
        "# spark_version = 'spark-3.3.0'\n",
        "\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",
        "# Install Postgres Driver\n",
        "!wget https://jdbc.postgresql.org/download/postgresql-42.5.0.jar\n",
        "\n",
        "# Set Environment Variables\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()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "\r0% [Working]\r            \rHit:1 https://cloud.r-project.org/bin/linux/ubuntu bionic-cran40/ InRelease\n",
            "\r0% [Connecting to archive.ubuntu.com] [Connecting to security.ubuntu.com (185.1\r0% [1 InRelease gpgv 3,626 B] [Connecting to archive.ubuntu.com] [Connecting to\r                                                                               \rIgn:2 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  InRelease\n",
            "\r0% [1 InRelease gpgv 3,626 B] [Connecting to archive.ubuntu.com] [Connecting to\r                                                                               \rHit:3 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  InRelease\n",
            "Hit:4 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  Release\n",
            "Get:5 http://security.ubuntu.com/ubuntu bionic-security InRelease [88.7 kB]\n",
            "Hit:6 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic InRelease\n",
            "Hit:7 http://archive.ubuntu.com/ubuntu bionic InRelease\n",
            "Get:8 http://archive.ubuntu.com/ubuntu bionic-updates InRelease [88.7 kB]\n",
            "Hit:9 http://ppa.launchpad.net/cran/libgit2/ubuntu bionic InRelease\n",
            "Hit:11 http://ppa.launchpad.net/deadsnakes/ppa/ubuntu bionic InRelease\n",
            "Get:12 http://archive.ubuntu.com/ubuntu bionic-backports InRelease [83.3 kB]\n",
            "Hit:13 http://ppa.launchpad.net/graphics-drivers/ppa/ubuntu bionic InRelease\n",
            "Fetched 261 kB in 3s (99.0 kB/s)\n",
            "Reading package lists... Done\n",
            "--2022-10-02 00:00:12--  https://jdbc.postgresql.org/download/postgresql-42.5.0.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: 1046274 (1022K) [application/java-archive]\n",
            "Saving to: ‘postgresql-42.5.0.jar.2’\n",
            "\n",
            "postgresql-42.5.0.j 100%[===================>]   1022K  5.11MB/s    in 0.2s    \n",
            "\n",
            "2022-10-02 00:00:13 (5.11 MB/s) - ‘postgresql-42.5.0.jar.2’ saved [1046274/1046274]\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "MMqDAjVS0KN9"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "spark = SparkSession.builder.appName(\"AmazonReview\").config(\"spark.driver.extraClassPath\",\"/content/postgresql-42.5.0.jar\").getOrCreate()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cyBsySGuY-9V"
      },
      "source": [
        "### Load Amazon Data into Spark DataFrame"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "CtCmBhQJY-9Z",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "6747e847-3b60-413f-b223-27f40f04080a"
      },
      "source": [
        "from pyspark import SparkFiles\n",
        "url = \"https://s3.amazonaws.com/amazon-reviews-pds/tsv/amazon_reviews_us_Digital_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_Digital_Video_Games_v1_00.tsv.gz\"), sep=\"\\t\", header=True, inferSchema=True)\n",
        "df.show()"
      ],
      "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|   21269168| RSH1OZ87OYK92|B013PURRZW|     603406193|Madden NFL 16 - X...|Digital_Video_Games|          2|            2|          3|   N|                N|A slight improvem...|I keep buying mad...|2015-08-31 00:00:00|\n",
            "|         US|     133437|R1WFOQ3N9BO65I|B00F4CEHNK|     341969535| Xbox Live Gift Card|Digital_Video_Games|          5|            0|          0|   N|                Y|          Five Stars|             Awesome|2015-08-31 00:00:00|\n",
            "|         US|   45765011| R3YOOS71KM5M9|B00DNHLFQA|     951665344|Command & Conquer...|Digital_Video_Games|          5|            0|          0|   N|                Y|Hail to the great...|If you are preppi...|2015-08-31 00:00:00|\n",
            "|         US|     113118|R3R14UATT3OUFU|B004RMK5QG|     395682204|Playstation Plus ...|Digital_Video_Games|          5|            0|          0|   N|                Y|          Five Stars|             Perfect|2015-08-31 00:00:00|\n",
            "|         US|   22151364| RV2W9SGDNQA2C|B00G9BNLQE|     640460561|Saints Row IV - E...|Digital_Video_Games|          5|            0|          0|   N|                Y|          Five Stars|            Awesome!|2015-08-31 00:00:00|\n",
            "|         US|   22151364|R3CFKLIZ0I2KOB|B00IMIL498|     621922192|Double Dragon: Ne...|Digital_Video_Games|          5|            0|          0|   N|                Y|          Five Stars|            Awesome!|2015-08-31 00:00:00|\n",
            "|         US|   38426028|R1LRYU1V0T3O38|B00S00IJH8|     215163395|              Sims 4|Digital_Video_Games|          4|            0|          0|   N|                Y|i like the new sk...|i like the new sk...|2015-08-31 00:00:00|\n",
            "|         US|    6057518| R44QKV6FE5CJ2|B004RMK4BC|     384246568|Playstation Netwo...|Digital_Video_Games|          5|            0|          0|   N|                Y|          Five Stars|               Super|2015-08-31 00:00:00|\n",
            "|         US|   20715661|R2TX1KLPXXXNYS|B00K59HKIQ|     384246568|Playstation Netwo...|Digital_Video_Games|          5|            0|          0|   N|                Y|         Easy & Fast|Excellent, fast a...|2015-08-31 00:00:00|\n",
            "|         US|   26540306|R1JEEW4C6R89BA|B00K59HKIQ|     384246568|Playstation Netwo...|Digital_Video_Games|          5|            0|          0|   N|                Y|          Five Stars|                  Ok|2015-08-31 00:00:00|\n",
            "|         US|    8926809|R3B3UHK1FO0ERS|B004774IPU|     151985175|Sid Meier's Civil...|Digital_Video_Games|          1|            0|          0|   N|                N|I am still playin...|As has been writt...|2015-08-31 00:00:00|\n",
            "|         US|   31525534|R2GVSDHW513SS1|B002LIT9EC|     695277014|Build-a-lot 4: Po...|Digital_Video_Games|          5|            0|          0|   N|                Y|Probably the best...|Probably the best...|2015-08-31 00:00:00|\n",
            "|         US|   22151364|R1R1NT516PYT73|B008ALUBYQ|     112160022|       Borderlands 2|Digital_Video_Games|          5|            0|          0|   N|                Y|          Five Stars|            Awesome!|2015-08-31 00:00:00|\n",
            "|         US|   22977584|R3K624QDQKENN9|B010KYDNDG|     835376637|Minecraft for PC/...|Digital_Video_Games|          4|            0|          0|   N|                Y|                 FUN|COOL BUT IT LAGES...|2015-08-31 00:00:00|\n",
            "|         US|   45765011| R1FOXH7PCJX3V|B008ALUBYQ|     112160022|       Borderlands 2|Digital_Video_Games|          1|            0|          2|   N|                Y|            One Star|Lames purchase I ...|2015-08-31 00:00:00|\n",
            "|         US|    2239522| RA1246M1OMDWC|B004RMK4P8|     384246568|Playstation Netwo...|Digital_Video_Games|          5|            0|          0|   N|                Y|          Five Stars|               Great|2015-08-31 00:00:00|\n",
            "|         US|   48805811|R2I9SXWB0PAEKQ|B004RMK5QG|     395682204|Playstation Plus ...|Digital_Video_Games|          5|            0|          0|   N|                Y|          Five Stars|          Awesome!!!|2015-08-31 00:00:00|\n",
            "|         US|   18646481|R3UGL544NA0G9C|B00BI16Z22|     552981447|Brink of Consciou...|Digital_Video_Games|          4|            0|          0|   N|                Y|       worth playing|pretty good but n...|2015-08-31 00:00:00|\n",
            "|         US|   10310935|R1CBA4Y92GVAVM|B004VSTQ2A|     232803743|Xbox Live Subscri...|Digital_Video_Games|          5|            0|          0|   N|                Y|          Five Stars|What can I say......|2015-08-31 00:00:00|\n",
            "|         US|    5587610|R24NEKNR01VEHU|B00GAC1D2G|     384246568|Playstation Netwo...|Digital_Video_Games|          5|            0|          0|   N|                Y|        Just amazing|Very fast to rece...|2015-08-31 00:00:00|\n",
            "+-----------+-----------+--------------+----------+--------------+--------------------+-------------------+-----------+-------------+-----------+----+-----------------+--------------------+--------------------+-------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "2yUSe55VY-9t"
      },
      "source": [
        "### Create DataFrames to match tables"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "C8REmY1aY-9u"
      },
      "source": [
        "from pyspark.sql.functions import to_date\n",
        "# Read in the Review dataset as a DataFrame\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Create the customers_table DataFrame\n",
        "customers_df = df.groupby(\"customer_id\").agg({\"customer_id\": \"count\"})\n",
        "customers_df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "V8F5-xFH-cbX",
        "outputId": "ff2521e9-ef98-4b30-f866-fa26fdab2170"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-----------+------------------+\n",
            "|customer_id|count(customer_id)|\n",
            "+-----------+------------------+\n",
            "|   26079415|                 1|\n",
            "|   12521601|                 2|\n",
            "|    4593696|                 1|\n",
            "|    1468259|                 1|\n",
            "|   38173350|                 1|\n",
            "|    3953141|                 2|\n",
            "|   20598424|                 1|\n",
            "|   46625306|                 1|\n",
            "|   47160828|                 2|\n",
            "|    1337034|                 1|\n",
            "|   14231016|                 1|\n",
            "|   49785748|                 1|\n",
            "|     712525|                 1|\n",
            "|   35339694|                 1|\n",
            "|    8211293|                 1|\n",
            "|   41832251|                 1|\n",
            "|    9037272|                 1|\n",
            "|   52036002|                 1|\n",
            "|   49514578|                 1|\n",
            "|   22875949|                 1|\n",
            "+-----------+------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "B0TESUDRY-90",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "9f307003-ac4c-438c-e211-8a9f9a8a468b"
      },
      "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()"
      ],
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-----------+--------------+\n",
            "|customer_id|customer_count|\n",
            "+-----------+--------------+\n",
            "|   26079415|             1|\n",
            "|   12521601|             2|\n",
            "|    4593696|             1|\n",
            "|    1468259|             1|\n",
            "|   38173350|             1|\n",
            "|    3953141|             2|\n",
            "|   20598424|             1|\n",
            "|   46625306|             1|\n",
            "|   47160828|             2|\n",
            "|    1337034|             1|\n",
            "|   14231016|             1|\n",
            "|   49785748|             1|\n",
            "|     712525|             1|\n",
            "|   35339694|             1|\n",
            "|    8211293|             1|\n",
            "|   41832251|             1|\n",
            "|    9037272|             1|\n",
            "|   52036002|             1|\n",
            "|   49514578|             1|\n",
            "|   22875949|             1|\n",
            "+-----------+--------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "4FwXA6UvY-96",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "d248258f-8736-46e5-b292-0227150e30cf"
      },
      "source": [
        "# Create the products_table DataFrame and drop duplicates. \n",
        "products_df = df.select([\"product_id\", \"product_title\"]).drop_duplicates()\n",
        "products_df.show()"
      ],
      "execution_count": 13,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----------+--------------------+\n",
            "|product_id|       product_title|\n",
            "+----------+--------------------+\n",
            "|B001TK3VTC|Cubis Gold 2 [Dow...|\n",
            "|B002MUB0TG|Little Shop of Tr...|\n",
            "|B00DRO824C|X2 The Threat [On...|\n",
            "|B00CMDKNHI| 100% Hidden Objects|\n",
            "|B00CMSCTA2|Age of Empires II...|\n",
            "|B00GDHD3LW|1 Month Subscript...|\n",
            "|B00OM1VVJM|            NBa 2K15|\n",
            "|B005XOJ6NK|   Sonic Generations|\n",
            "|B00GGUHGGY|Call of Duty Blac...|\n",
            "|B00A9XQ890|Grand Theft Auto:...|\n",
            "|B003YMMGEE|Sniper: Ghost War...|\n",
            "|B00405RA04|Rome: Total War -...|\n",
            "|B0083KX9IO|Stronghold Kingdo...|\n",
            "|B00M0W1YZQ|Chronicles of Wit...|\n",
            "|B00895ARJC|  Syberia [Download]|\n",
            "|B0084FHQ1O|The Stronghold Co...|\n",
            "|B00KK9PR90|Freddi Fish 3: Th...|\n",
            "|B001KC01OW|Wonderlines [Down...|\n",
            "|B008TTPTOQ|Battlefield Bad C...|\n",
            "|B00DGZDGUM|Nascar The Game 2...|\n",
            "+----------+--------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "MkqyCuNQY-9-",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "80b49718-e071-4370-8c30-1d665904a9ac"
      },
      "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()"
      ],
      "execution_count": 14,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------------+-----------+----------+--------------+-----------+\n",
            "|     review_id|customer_id|product_id|product_parent|review_date|\n",
            "+--------------+-----------+----------+--------------+-----------+\n",
            "| RSH1OZ87OYK92|   21269168|B013PURRZW|     603406193| 2015-08-31|\n",
            "|R1WFOQ3N9BO65I|     133437|B00F4CEHNK|     341969535| 2015-08-31|\n",
            "| R3YOOS71KM5M9|   45765011|B00DNHLFQA|     951665344| 2015-08-31|\n",
            "|R3R14UATT3OUFU|     113118|B004RMK5QG|     395682204| 2015-08-31|\n",
            "| RV2W9SGDNQA2C|   22151364|B00G9BNLQE|     640460561| 2015-08-31|\n",
            "|R3CFKLIZ0I2KOB|   22151364|B00IMIL498|     621922192| 2015-08-31|\n",
            "|R1LRYU1V0T3O38|   38426028|B00S00IJH8|     215163395| 2015-08-31|\n",
            "| R44QKV6FE5CJ2|    6057518|B004RMK4BC|     384246568| 2015-08-31|\n",
            "|R2TX1KLPXXXNYS|   20715661|B00K59HKIQ|     384246568| 2015-08-31|\n",
            "|R1JEEW4C6R89BA|   26540306|B00K59HKIQ|     384246568| 2015-08-31|\n",
            "|R3B3UHK1FO0ERS|    8926809|B004774IPU|     151985175| 2015-08-31|\n",
            "|R2GVSDHW513SS1|   31525534|B002LIT9EC|     695277014| 2015-08-31|\n",
            "|R1R1NT516PYT73|   22151364|B008ALUBYQ|     112160022| 2015-08-31|\n",
            "|R3K624QDQKENN9|   22977584|B010KYDNDG|     835376637| 2015-08-31|\n",
            "| R1FOXH7PCJX3V|   45765011|B008ALUBYQ|     112160022| 2015-08-31|\n",
            "| RA1246M1OMDWC|    2239522|B004RMK4P8|     384246568| 2015-08-31|\n",
            "|R2I9SXWB0PAEKQ|   48805811|B004RMK5QG|     395682204| 2015-08-31|\n",
            "|R3UGL544NA0G9C|   18646481|B00BI16Z22|     552981447| 2015-08-31|\n",
            "|R1CBA4Y92GVAVM|   10310935|B004VSTQ2A|     232803743| 2015-08-31|\n",
            "|R24NEKNR01VEHU|    5587610|B00GAC1D2G|     384246568| 2015-08-31|\n",
            "+--------------+-----------+----------+--------------+-----------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "lzMmkdKmY--D",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "90872ccf-00a3-4888-e107-b7bdcb5ad0bc"
      },
      "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()"
      ],
      "execution_count": 15,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------------+-----------+-------------+-----------+----+-----------------+\n",
            "|     review_id|star_rating|helpful_votes|total_votes|vine|verified_purchase|\n",
            "+--------------+-----------+-------------+-----------+----+-----------------+\n",
            "| RSH1OZ87OYK92|          2|            2|          3|   N|                N|\n",
            "|R1WFOQ3N9BO65I|          5|            0|          0|   N|                Y|\n",
            "| R3YOOS71KM5M9|          5|            0|          0|   N|                Y|\n",
            "|R3R14UATT3OUFU|          5|            0|          0|   N|                Y|\n",
            "| RV2W9SGDNQA2C|          5|            0|          0|   N|                Y|\n",
            "|R3CFKLIZ0I2KOB|          5|            0|          0|   N|                Y|\n",
            "|R1LRYU1V0T3O38|          4|            0|          0|   N|                Y|\n",
            "| R44QKV6FE5CJ2|          5|            0|          0|   N|                Y|\n",
            "|R2TX1KLPXXXNYS|          5|            0|          0|   N|                Y|\n",
            "|R1JEEW4C6R89BA|          5|            0|          0|   N|                Y|\n",
            "|R3B3UHK1FO0ERS|          1|            0|          0|   N|                N|\n",
            "|R2GVSDHW513SS1|          5|            0|          0|   N|                Y|\n",
            "|R1R1NT516PYT73|          5|            0|          0|   N|                Y|\n",
            "|R3K624QDQKENN9|          4|            0|          0|   N|                Y|\n",
            "| R1FOXH7PCJX3V|          1|            0|          2|   N|                Y|\n",
            "| RA1246M1OMDWC|          5|            0|          0|   N|                Y|\n",
            "|R2I9SXWB0PAEKQ|          5|            0|          0|   N|                Y|\n",
            "|R3UGL544NA0G9C|          4|            0|          0|   N|                Y|\n",
            "|R1CBA4Y92GVAVM|          5|            0|          0|   N|                Y|\n",
            "|R24NEKNR01VEHU|          5|            0|          0|   N|                Y|\n",
            "+--------------+-----------+-------------+-----------+----+-----------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "jITZhLkmY--J"
      },
      "source": [
        "### Connect to the AWS RDS instance and write each DataFrame to its table. "
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "7jiUvs1aY--L",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "ea023638-6d6a-4995-f082-152093a2ad18"
      },
      "source": [
        "# Store environmental variable\n",
        "from getpass import getpass\n",
        "password = getpass('Enter database password')\n",
        "\n",
        "# Configure settings for RDS\n",
        "mode = \"append\"\n",
        "jdbc_url=\"jdbc:postgresql://amazonreviews.cpurzdxxzf4z.us-west-1.rds.amazonaws.com:5432/amazonReviews\"\n",
        "config = {\"user\":\"postgres\", \n",
        "          \"password\": password, \n",
        "          \"driver\":\"org.postgresql.Driver\"}"
      ],
      "execution_count": 16,
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "Enter database password··········\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "T2zgZ-aKY--Q",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 1000
        },
        "outputId": "0b6814c7-199b-4afb-ecfe-4deedb959d5b"
      },
      "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)"
      ],
      "execution_count": 17,
      "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 o117.jdbc.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 17.0 failed 1 times, most recent failure: Lost task 0.0 in stage 17.0 (TID 13) (229f3e900565 executor driver): java.sql.BatchUpdateException: Batch entry 0 INSERT INTO review_id_table (\"review_id\",\"customer_id\",\"product_id\",\"product_parent\",\"review_date\") VALUES ('RSH1OZ87OYK92',21269168,'B013PURRZW',603406193,'2015-08-31 +00'::date) was aborted: ERROR: duplicate key value violates unique constraint \"review_id_table_pkey\"\n  Detail: Key (review_id)=(RSH1OZ87OYK92) already exists.  Call getNextException to see other errors in the batch.\n\tat org.postgresql.jdbc.BatchResultHandler.handleError(BatchResultHandler.java:165)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2367)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2099)\n\tat org.postgresql.core.v3.QueryExecutorImpl.flushIfDeadlockRisk(QueryExecutorImpl.java:1456)\n\tat org.postgresql.core.v3.QueryExecutorImpl.sendQuery(QueryExecutorImpl.java:1481)\n\tat org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:546)\n\tat org.postgresql.jdbc.PgStatement.internalExecuteBatch(PgStatement.java:893)\n\tat org.postgresql.jdbc.PgStatement.executeBatch(PgStatement.java:916)\n\tat org.postgresql.jdbc.PgPreparedStatement.executeBatch(PgPreparedStatement.java:1684)\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)=(RSH1OZ87OYK92) already exists.\n\tat org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2676)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2366)\n\t... 21 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 ('RSH1OZ87OYK92',21269168,'B013PURRZW',603406193,'2015-08-31 +00'::date) was aborted: ERROR: duplicate key value violates unique constraint \"review_id_table_pkey\"\n  Detail: Key (review_id)=(RSH1OZ87OYK92) already exists.  Call getNextException to see other errors in the batch.\n\tat org.postgresql.jdbc.BatchResultHandler.handleError(BatchResultHandler.java:165)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2367)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2099)\n\tat org.postgresql.core.v3.QueryExecutorImpl.flushIfDeadlockRisk(QueryExecutorImpl.java:1456)\n\tat org.postgresql.core.v3.QueryExecutorImpl.sendQuery(QueryExecutorImpl.java:1481)\n\tat org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:546)\n\tat org.postgresql.jdbc.PgStatement.internalExecuteBatch(PgStatement.java:893)\n\tat org.postgresql.jdbc.PgStatement.executeBatch(PgStatement.java:916)\n\tat org.postgresql.jdbc.PgPreparedStatement.executeBatch(PgPreparedStatement.java:1684)\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)=(RSH1OZ87OYK92) already exists.\n\tat org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2676)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2366)\n\t... 21 more\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "1m3yzn-LY--U",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 1000
        },
        "outputId": "6c05bc05-f17a-423e-c3ec-66ed31865297"
      },
      "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)"
      ],
      "execution_count": 18,
      "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-18-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.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 o123.jdbc.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 20.0 failed 1 times, most recent failure: Lost task 0.0 in stage 20.0 (TID 15) (229f3e900565 executor driver): java.sql.BatchUpdateException: Batch entry 0 INSERT INTO products_table (\"product_id\",\"product_title\") VALUES ('B001TK3VTC','Cubis Gold 2 [Download]') was aborted: ERROR: duplicate key value violates unique constraint \"products_table_pkey\"\n  Detail: Key (product_id)=(B001TK3VTC) already exists.  Call getNextException to see other errors in the batch.\n\tat org.postgresql.jdbc.BatchResultHandler.handleError(BatchResultHandler.java:165)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2367)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2099)\n\tat org.postgresql.core.v3.QueryExecutorImpl.flushIfDeadlockRisk(QueryExecutorImpl.java:1456)\n\tat org.postgresql.core.v3.QueryExecutorImpl.sendQuery(QueryExecutorImpl.java:1481)\n\tat org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:546)\n\tat org.postgresql.jdbc.PgStatement.internalExecuteBatch(PgStatement.java:893)\n\tat org.postgresql.jdbc.PgStatement.executeBatch(PgStatement.java:916)\n\tat org.postgresql.jdbc.PgPreparedStatement.executeBatch(PgPreparedStatement.java:1684)\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 \"products_table_pkey\"\n  Detail: Key (product_id)=(B001TK3VTC) already exists.\n\tat org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2676)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2366)\n\t... 21 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 products_table (\"product_id\",\"product_title\") VALUES ('B001TK3VTC','Cubis Gold 2 [Download]') was aborted: ERROR: duplicate key value violates unique constraint \"products_table_pkey\"\n  Detail: Key (product_id)=(B001TK3VTC) already exists.  Call getNextException to see other errors in the batch.\n\tat org.postgresql.jdbc.BatchResultHandler.handleError(BatchResultHandler.java:165)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2367)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2099)\n\tat org.postgresql.core.v3.QueryExecutorImpl.flushIfDeadlockRisk(QueryExecutorImpl.java:1456)\n\tat org.postgresql.core.v3.QueryExecutorImpl.sendQuery(QueryExecutorImpl.java:1481)\n\tat org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:546)\n\tat org.postgresql.jdbc.PgStatement.internalExecuteBatch(PgStatement.java:893)\n\tat org.postgresql.jdbc.PgStatement.executeBatch(PgStatement.java:916)\n\tat org.postgresql.jdbc.PgPreparedStatement.executeBatch(PgPreparedStatement.java:1684)\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 \"products_table_pkey\"\n  Detail: Key (product_id)=(B001TK3VTC) already exists.\n\tat org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2676)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2366)\n\t... 21 more\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "KbXri15fY--Z"
      },
      "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)"
      ],
      "execution_count": 19,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "XdQknSHLY--e"
      },
      "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)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Exuo6ebUsCqW"
      },
      "source": [],
      "execution_count": null,
      "outputs": []
    }
  ]
}