<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_Sports_v1_00.tsv.gz"
spark.sparkContext.addFile(url)
sports_df = spark.read.csv(SparkFiles.get("amazon_reviews_us_Sports_v1_00.tsv.gz"), sep="\t", header=True, inferSchema=True)
sports_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|   48945260|R1WBPB8MDCCN8F|B012P7UPSM|     409940130|Chicago Blackhawk...|          Sports|          5|            0|          0|   N|                N|   LOVE IT. 6 stars!|Bought this last ...| 2015-08-31|
|         US|    5782091|R32M0YEWV77XG8|B001GQ3VHG|     657746679|Copag Poker Size ...|          Sports|          5|    [...]<jupyter_text>Size of Data<jupyter_code>sports_df.count()<jupyter_output><empty_output><jupyter_text>Cleaned up DataFrames to match tables<jupyter_code>from pyspark.sql.functions import to_date
review_id_df = sports_df.select(["review_id", "customer_id", "product_id", "product_parent", to_date("review_date", 'yyyy-MM-dd').alias("review_date")])
review_id_df.show()
products_df = sports_df.select(["product_id", "product_title"]).dropDuplicates(["product_id"])
products_df.show(20)
customers_df = sports_df.groupby("customer_id").agg({"customer_id": "count"}).withColumnRenamed("count(customer_id)", "customer_count")
customers_df.show()
vine_df = sports_df.select(["review_id", "star_rating", "helpful_votes", "total_votes", "vine"])
vine_df= vine_df.withColumn("star_rating", vine_df.star_rating.cast("int"))
vine_df.show(10)<jupyter_output>+--------------+-----------+-------------+-----------+----+
|     review_id|star_rating|helpful_votes|total_votes|vine|
+--------------+-----------+-------------+-----------+----+
|R1WBPB8MDCCN8F|          5|            0|          0|   N|
|R32M0YEWV77XG8|          5|            1|          1|   N|
| RR8V7WR27NXJ5|          1|            0|          0|   N|
|R1MHO5V9Z932AY|          5|            0|          0|   N|
|R16PD71086BD2V|          5|            0|          1|   N|
|R1Z8IFGWTRWXT6|          3|            0|          0|   N|
|R3AUMSHAW73HWN|          4|            2|          3|   N|
|R2KWDWFOHGX6FL|          5|            0|          0|   N|
|R3H9543FWBWFBU|          5|            1|          1|   N|
| RUANXOQ9W3OU5|          5|            0|          0|   N|
+--------------+-----------+-------------+-----------+----+
only showing top 10 rows<jupyter_text>Push to AWS RDS instance<jupyter_code>mode = "append"
jdbc_url="jdbc:postgresql://<Insert Endpoint>:5432/big-data-challenge"
config = {"user":"postgres", "password": "<Insert 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
products_df.write.jdbc(url=jdbc_url, table='products', mode=mode, properties=config)
# Write customers_df to table in RDS
customers_df.write.jdbc(url=jdbc_url, table='customers', mode=mode, properties=config)
# Write vine_df to table in RDS
vine_df.write.jdbc(url=jdbc_url, table='vine_table', mode=mode, properties=config)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.2680652680652681, 'fraction_numerical': 0.039886039886039885, 'mean_word_length': 1.9845440494590418, 'pattern_counts': {'":': 4, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '519372', 'license': 'ADSL', 'n_tokens_mistral': 1354, 'n_tokens_neox': 1240, 'n_words': 194, 'path': 'level_one.ipynb', 'repo_name': 'jtwilliams17/big-data-challenge', 'size': '3861'}