<jupyter_start><jupyter_text>Load Amazon Data into Spark DataFrame<jupyter_code>from pyspark import SparkFiles
url = "https://s3.amazonaws.com/amazon-reviews-pds/tsv/amazon_reviews_us_Musical_Instruments_v1_00.tsv.gz"
spark.sparkContext.addFile(url)
df = spark.read.option("encoding", "UTF-8").csv(SparkFiles.get(""), sep="\t", header=True, inferSchema=True)
df.show()<jupyter_output>+-----------+-----------+--------------+----------+--------------+--------------------+-------------------+-----------+-------------+-----------+----+-----------------+--------------------+--------------------+-----------+
|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|
+-----------+-----------+--------------+----------+--------------+--------------------+-------------------+-----------+-------------+-----------+----+-----------------+--------------------+--------------------+-----------+
|         US|   45610553| RMDCHWD0Y5OZ9|B00HH62VB6|     618218723|AGPtek® 10 Isolat...|Musical Instruments|          3|            0|          1|   N|                N|         Three Stars|Works very good, ...| 2015-08-31|
|         US|   14640079| RZSL0BALIYUNU|B003LRN53I|     986692292|Sennheiser HD203 ...|Musical Instruments| [...]<jupyter_text>Create DataFrames to match tables<jupyter_code>from pyspark.sql.functions import count
from pyspark.sql.functions import sum
from pyspark.sql.functions import to_date
# Read in the Review dataset as a DataFrame
customers_df = df

# Create the customers_table DataFrame
customers_df = df.groupBy("customer_id").agg(count("customer_id").alias("customer_count"))
# Resource - https://www.geeksforgeeks.org/how-to-name-aggregate-columns-in-pyspark-dataframe/
customers_df.show()
# Create the products_table DataFrame and drop duplicates. 
products_df = df.select(["product_id", "product_title"]).drop_duplicates()
products_df.show()

# Create the review_id_table DataFrame. 
# Convert the 'review_date' column to a date datatype with to_date("review_date", 'yyyy-MM-dd').alias("review_date")
review_id_df = df.select(['review_id', 'customer_id','product_id', 'product_parent',  to_date("review_date", 'yyyy-MM-dd').alias("review_date")])
review_id_df.show()
# Create the vine_table. DataFrame
vine_df = df.select(['review_id', 'star_rating', 'helpful_votes','total_votes', 'vine', 'verified_purchase'])
vine_df.show()<jupyter_output>+--------------+-----------+-------------+-----------+----+-----------------+
|     review_id|star_rating|helpful_votes|total_votes|vine|verified_purchase|
+--------------+-----------+-------------+-----------+----+-----------------+
| RMDCHWD0Y5OZ9|          3|            0|          1|   N|                N|
| RZSL0BALIYUNU|          5|            0|          0|   N|                Y|
| RIZR67JKUDBI0|          3|            0|          1|   N|                Y|
|R27HL570VNL85F|          5|            0|          0|   N|                Y|
|R34EBU9QDWJ1GD|          5|            0|          0|   N|                Y|
|R1WCUI4Z1SIQEO|          5|            0|          0|   N|                N|
| RL5LNO26GAVJ1|          2|            3|          4|   N|                Y|
|R3GYQ5W8JHP8SB|          5|            0|          0|   N|                Y|
|R30SHYQXGG5EYC|          5|            0|          0|   N|                Y|
|R14YLXA56NP51I|          5|            1|          1|   N|     [...]<jupyter_text>Connect to the AWS RDS instance and write each DataFrame to its table.<jupyter_code># Configure settings for RDS
mode = "append"
jdbc_url="jdbc:postgresql://<REDACTED>/<REDACTED."
config = {"user":"REDACTED", 
          "password": "<PASSWORD>", 
          "driver":"org.postgresql.Driver"}
# Write review_id_df to table in RDS
review_id_df.write.jdbc(url=jdbc_url, table='review_id_table', mode=mode, properties=config)
# Write products_df to table in RDS
# about 3 min
products_df.write.jdbc(url=jdbc_url, table='products_table', mode=mode, properties=config)
# Write customers_df to table in RDS
# 5 min 14 s
customers_df.write.jdbc(url=jdbc_url, table='customers_table', mode=mode, properties=config)
# Write vine_df to table in RDS
# 11 minutes
vine_df.write.jdbc(url=jdbc_url, table='vine_table', mode=mode, properties=config)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.23929299796057105, 'fraction_numerical': 0.032404260140493996, 'mean_word_length': 1.8925294888597641, 'pattern_counts': {'":': 3, '<': 14, '<?xml version=': 0, '>': 13, 'https://': 2, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '655546', 'license': 'ADSL', 'n_tokens_mistral': 1481, 'n_tokens_neox': 1354, 'n_words': 262, 'path': 'Amazon_Reviews_ETL.ipynb', 'repo_name': 'LleeMcD/Amazon_Vine_Analysis', 'size': '4411'}