<jupyter_start><jupyter_text>Dependencies<jupyter_code>import os
# Find the latest version of spark 3.0  from http://www-us.apache.org/dist/spark/ and enter as the spark version
# For example:
spark_version = 'spark-3.0.1'
# spark_version = 'spark-3.<spark version>'
os.environ['SPARK_VERSION'] = spark_version

# Install Spark and Java
!apt-get update
!apt-get install openjdk-11-jdk-headless -qq > /dev/null
!wget -q https://archive.apache.org/dist/spark/$SPARK_VERSION/$SPARK_VERSION-bin-hadoop2.7.tgz
!tar xf $SPARK_VERSION-bin-hadoop2.7.tgz
!pip install -q findspark

# Set Environment Variables
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-11-openjdk-amd64"
os.environ["SPARK_HOME"] = f"/content/{spark_version}-bin-hadoop2.7"

# Start a SparkSession
import findspark
findspark.init()

!wget https://jdbc.postgresql.org/download/postgresql-42.3.3.jar
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("BigData-HW-1").config("spark.driver.extraClassPath","/content/postgresql-42.2.9.jar").getOrCreate()<jupyter_output><empty_output><jupyter_text>Amazon Data<jupyter_code>from pyspark import SparkFiles

# load data
url = "https://s3.amazonaws.com/amazon-reviews-pds/tsv/amazon_reviews_us_Video_Games_v1_00.tsv.gz"
spark.sparkContext.addFile(url)
video_games_df = spark.read.csv(SparkFiles.get("amazon_reviews_us_Video_Games_v1_00.tsv.gz"), sep="\t", header=True, inferSchema=True)
video_games_df.show()
# clean dataframe
from pyspark.sql.functions import to_date
# Review DataFrame
review_id_df = video_games_df.select(["review_id",
                                      "customer_id",
                                      "product_id",
                                      "product_parent",
                                      to_date("review_date", 'yyyy-MM-dd').alias("review_date")])
review_id_df.show()
# remove product duplicates
products_df = video_games_df.select(["product_id", "product_title"]).drop_duplicates()
products_df.show()
# select relevent reviews data
reviews_df = video_games_df.select(["review_id", "review_headline", "review_body"])
reviews_df.show()
# clean customer data
customers_df = video_games_df.groupby("customer_id").agg({"customer_id": "count"}).withColumnRenamed("count(customer_id)", "customer_count")
customers_df.show()
# select relevent vine videos data
vine_df = video_games_df.select(["review_id", "star_rating", "helpful_votes", "total_votes", "vine"])
vine_df.show(10)<jupyter_output>+--------------+-----------+-------------+-----------+----+
|     review_id|star_rating|helpful_votes|total_votes|vine|
+--------------+-----------+-------------+-----------+----+
| RTIS3L2M1F5SM|          5|            0|          0|   N|
| R1ZV7R40OLHKD|          5|            0|          0|   N|
|R3BH071QLH8QMC|          1|            0|          1|   N|
|R127K9NTSXA2YH|          3|            0|          0|   N|
|R32ZWUXDJPW27Q|          4|            0|          0|   N|
|R3AQQ4YUKJWBA6|          1|            0|          0|   N|
|R2F0POU5K6F73F|          5|            0|          0|   N|
|R3VNR804HYSMR6|          5|            0|          0|   N|
| R3GZTM72WA2QH|          5|            0|          0|   N|
| RNQOY62705W1K|          4|            0|          0|   N|
+--------------+-----------+-------------+-----------+----+
only showing top 10 rows<jupyter_text>Push to RDS<jupyter_code># setup
mode = "append"
jdbc_url="jdbc:postgresql://rds-postgresql-bigdatadb.cin48putgxh5.us-east-1.rds.amazonaws.com:5432/bigdataDB"
config = {"user":"epurwant0", "password": "<PASSWORD>", "driver":"org.postgresql.Driver"}
# write tables to RDS
review_id_df.write.jdbc(url=jdbc_url, table='review_id_table', mode=mode, properties=config)
products_df.write.jdbc(url=jdbc_url, table='products', mode=mode, properties=config)
customers_df.write.jdbc(url=jdbc_url, table='customers', mode=mode, properties=config)
vine_df.write.jdbc(url=jdbc_url, table='vines', mode=mode, properties=config)<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup