# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     name: python3
# ---

# + colab={"base_uri": "https://localhost:8080/"} id="SDU1ACR6A4L0" executionInfo={"status": "ok", "timestamp": 1638413046417, "user_tz": 480, "elapsed": 24024, "user": {"displayName": "<NAME>", "photoUrl": "https://lh3.googleusercontent.com/a/default-user=s64", "userId": "02514073448309360592"}} outputId="903d932f-8536-4e8f-9e37-5c16158119c6"
import os
# Find the latest version of spark 3.0  from http://www.apache.org/dist/spark/ and enter as the spark version
# For example:
# spark_version = 'spark-3.0.1'
spark_version = 'spark-3.0.3'
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 http://www.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()

# + colab={"base_uri": "https://localhost:8080/"} id="S3BQnXYHBABd" executionInfo={"status": "ok", "timestamp": 1638413047413, "user_tz": 480, "elapsed": 631, "user": {"displayName": "<NAME>", "photoUrl": "https://lh3.googleusercontent.com/a/default-user=s64", "userId": "02514073448309360592"}} outputId="11c9ac79-a773-4707-f9a0-b676cb41c155"
# !wget https://jdbc.postgresql.org/download/postgresql-42.2.9.jar


# + id="oxJLmVv7BAGj" executionInfo={"status": "ok", "timestamp": 1638413211279, "user_tz": 480, "elapsed": 7238, "user": {"displayName": "<NAME>", "photoUrl": "https://lh3.googleusercontent.com/a/default-user=s64", "userId": "02514073448309360592"}}
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("BigData-HW-Part-1").config("spark.driver.extraClassPath", "/content.postgressql-42.2.9.jar").getOrCreate()

# + colab={"base_uri": "https://localhost:8080/"} id="7bjaea1UBAKj" executionInfo={"status": "ok", "timestamp": 1638413423791, "user_tz": 480, "elapsed": 35279, "user": {"displayName": "<NAME>", "photoUrl": "https://lh3.googleusercontent.com/a/default-user=s64", "userId": "02514073448309360592"}} outputId="800d1357-a452-4d7e-90be-4611232f86c1"
#Load Amazon data into Spark DataFrame
from pyspark import SparkFiles
url = 'https://s3.amazonaws.com/amazon-reviews-pds/tsv/amazon_reviews_us_Video_Games_v1_00.tsv.gz'

spark.sparkContext.addFile(url)
video_game_df = spark.read.csv(SparkFiles.get("amazon_reviews_us_Video_Games_v1_00.tsv.gz"), sep='\t', header=True, inferSchema=True)
video_game_df.show()

# + colab={"base_uri": "https://localhost:8080/"} id="51KgqjVKCY50" executionInfo={"status": "ok", "timestamp": 1638413464850, "user_tz": 480, "elapsed": 10583, "user": {"displayName": "<NAME>", "photoUrl": "https://lh3.googleusercontent.com/a/default-user=s64", "userId": "02514073448309360592"}} outputId="e14caf59-b38c-42bb-a9af-78fd28465b6f"
#Size of Data
video_game_df.count()



# + colab={"base_uri": "https://localhost:8080/"} id="D28D3nDwCY8j" executionInfo={"status": "ok", "timestamp": 1638413839013, "user_tz": 480, "elapsed": 403, "user": {"displayName": "<NAME>", "photoUrl": "https://lh3.googleusercontent.com/a/default-user=s64", "userId": "02514073448309360592"}} outputId="4d6ff147-5e03-4336-def7-a7bf00d08e11"
#Clean DataFrame to match tables
from pyspark.sql.functions import to_date

review_id_df = video_game_df.select(['review_id', 'customer_id', 'product_id', 'product_parent', to_date('review_date', 'yyyy-MM-dd').alias('review_date')])
review_id_df.show()

# + id="QBSoYE3OCY_O" executionInfo={"status": "ok", "timestamp": 1638413970284, "user_tz": 480, "elapsed": 124, "user": {"displayName": "<NAME>", "photoUrl": "https://lh3.googleusercontent.com/a/default-user=s64", "userId": "02514073448309360592"}}
products_df = video_game_df.select(['product_id', 'product_title']).drop_duplicates()

# + colab={"base_uri": "https://localhost:8080/"} id="zqQrOeghE3at" executionInfo={"status": "ok", "timestamp": 1638413992360, "user_tz": 480, "elapsed": 15699, "user": {"displayName": "<NAME>", "photoUrl": "https://lh3.googleusercontent.com/a/default-user=s64", "userId": "02514073448309360592"}} outputId="babff7fa-a569-4a96-975f-d71400327a48"
products_df.show()

# + colab={"base_uri": "https://localhost:8080/"} id="Atc79p_rE3dj" executionInfo={"status": "ok", "timestamp": 1638414034337, "user_tz": 480, "elapsed": 450, "user": {"displayName": "<NAME>", "photoUrl": "https://lh3.googleusercontent.com/a/default-user=s64", "userId": "02514073448309360592"}} outputId="776debd8-19e0-449d-aa61-68451d9ade35"
reviews_df = video_game_df.select(['review_id', 'review_headline', 'review_body'])
reviews_df.show(10)

# + colab={"base_uri": "https://localhost:8080/"} id="mBlducDBE3f1" executionInfo={"status": "ok", "timestamp": 1638414122638, "user_tz": 480, "elapsed": 16597, "user": {"displayName": "<NAME>", "photoUrl": "https://lh3.googleusercontent.com/a/default-user=s64", "userId": "02514073448309360592"}} outputId="3f0f8536-a271-4984-b7ef-da8be5262d6a"
customers_df = video_game_df.groupBy('customer_id').agg({'customer_id': 'count'}).withColumnRenamed('count(customer_id', 'customer_count')
customers_df.show()

# + colab={"base_uri": "https://localhost:8080/"} id="u6UtxDo_Fi4M" executionInfo={"status": "ok", "timestamp": 1638414167617, "user_tz": 480, "elapsed": 290, "user": {"displayName": "<NAME>", "photoUrl": "https://lh3.googleusercontent.com/a/default-user=s64", "userId": "02514073448309360592"}} outputId="f336af08-f4c7-404a-92d0-4f93b05de6a8"
vine_df = video_game_df.select(['review_id', 'star_rating', 'helpful_votes', 'total_votes', 'vine'])
vine_df.show(10)

# + id="RkeF5kP_Kjgz"
#Schema
''''CREATE TABLE review_id_table (
	review_id INT,
	customer_id TEXT,
	product_id TEXT,
	product_parent TEXT,
	review_date DATE
	
);

CREATE TABLE product (
	product_id TEXT,
	product_title TEXT
);

CREATE TABLE customers (
	customer_id INT
	customer_count INT

);

CREATE TABLE vine_table (
	review_id TEXT,
	star_rating INT,
	helpful_votes INT,
	total_votes INT,
	vine TEXT
	

);

'''

# + id="5dl9JRp7Fi6y" executionInfo={"status": "ok", "timestamp": 1638414723426, "user_tz": 480, "elapsed": 126, "user": {"displayName": "<NAME>", "photoUrl": "https://lh3.googleusercontent.com/a/default-user=s64", "userId": "02514073448309360592"}}
#Push to AWS RDS instance
mode = 'append'
jdbc_url = ''
config = {'user':'', 'password': '', 'driver': 'org.postgresql.Driver'}


# + id="Nu4BU9k9Fi9T" executionInfo={"status": "ok", "timestamp": 1638416259261, "user_tz": 480, "elapsed": 3, "user": {"displayName": "<NAME>", "photoUrl": "https://lh3.googleusercontent.com/a/default-user=s64", "userId": "02514073448309360592"}}
#write review_id to table in RDS
review_id_df.write.jdbc(url=jdbc_url, table='review_id_table', mode=mode, properties=config)

# + id="LWJbxSY7Jnu8"


# + id="wAMAIJ0KFi_p"


# + id="CPu4TqUNCZBd"


# + id="9jCaNMD8BANM"


StarcoderdataJupyterScriptsDedupFiltered