<jupyter_start><jupyter_text><jupyter_code># Connect to Postgres 
!wget https://jdbc.postgresql.org/download/postgresql-42.2.9.jar<jupyter_output>--2022-01-27 03:22:40--  https://jdbc.postgresql.org/download/postgresql-42.2.9.jar
Resolving jdbc.postgresql.org (jdbc.postgresql.org)... 192.168.3.11, fdf8:f53e:61e4::18
Connecting to jdbc.postgresql.org (jdbc.postgresql.org)|192.168.3.11|:443... connected.
HTTP request sent, awaiting response... 200 OK
Length: 914037 (893K) [application/java-archive]
Saving to: ‘postgresql-42.2.9.jar.1’

postgresql-42.2.9.j 100%[===================>] 892.61K  5.16MB/s    in 0.2s    

2022-01-27 03:22:40 (5.16 MB/s) - ‘postgresql-42.2.9.jar.1’ saved [914037/914037]<jupyter_text>Extraction<jupyter_code># Start Spark session
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("CloudETLProjectAnalysis").config("spark.driver.extraClassPath","/content/postgresql-42.2.9.jar").getOrCreate()
from pyspark import SparkFiles

# Importing electronic reviews
url = "https://s3.amazonaws.com/amazon-reviews-pds/tsv/amazon_reviews_us_Electronics_v1_00.tsv.gz"
spark.sparkContext.addFile(url)

elec_df = spark.read.option('header', 'true').csv(SparkFiles.get("amazon_reviews_us_Electronics_v1_00.tsv.gz"), inferSchema=True, sep='\t', timestampFormat="mm/dd/yy")
elec_df.show(10)
# Importing sports reviews
url = "https://s3.amazonaws.com/amazon-reviews-pds/tsv/amazon_reviews_us_Sports_v1_00.tsv.gz"
spark.sparkContext.addFile(url)

sport_df = spark.read.option('header', 'true').csv(SparkFiles.get("amazon_reviews_us_Sports_v1_00.tsv.gz"), inferSchema=True, sep='\t', timestampFormat="mm/dd/yy")
sport_df.show(10)
# Concatinate two dataframes 
df = elec_df.union(sport_df)
df.show(10)
# Count number of rows
df.count()
# Drop any nulls
df = df.dropna()

# Check the shape of the dataframe
print((df.count(), len(df.columns)))
# Check data types
df.printSchema()<jupyter_output>root
 |-- marketplace: string (nullable = true)
 |-- customer_id: integer (nullable = true)
 |-- review_id: string (nullable = true)
 |-- product_id: string (nullable = true)
 |-- product_parent: integer (nullable = true)
 |-- product_title: string (nullable = true)
 |-- product_category: string (nullable = true)
 |-- star_rating: string (nullable = true)
 |-- helpful_votes: integer (nullable = true)
 |-- total_votes: integer (nullable = true)
 |-- vine: string (nullable = true)
 |-- verified_purchase: string (nullable = true)
 |-- review_headline: string (nullable = true)
 |-- review_body: string (nullable = true)
 |-- review_date: string (nullable = true)<jupyter_text>Transformation<jupyter_code># to fit review_id_table
review_id_df = df.select('review_id', 'customer_id', 'product_id', 'product_parent', 'review_date')
review_id_df.show()
# Check for duplicates
review_id_df.groupby(['review_id']).count().where('count > 1').sort('count', ascending=False).show()
# Check data types
review_id_df.printSchema()
# to fit products table
products_df = df.select('product_id', 'product_title')
products_df.show()
# Check for duplicates
products_df.groupby(['product_id']).count().where('count > 1').sort('count', ascending=False).show()
# Drop duplicates
products_df = products_df.dropDuplicates(['product_id'])
# Check data types
products_df.printSchema()
products_df.count()
# to fit customers table
customers_df = df.select('customer_id')
customers_df = customers_df.groupBy('customer_id').count()
customers_df = customers_df.withColumnRenamed('count', 'customer_count')
customers_df.show()
# Check data types
customers_df.printSchema()
customers_df.count()
# to fit vine table
vine_df = df.select('review_id', 'star_rating', 'helpful_votes', 'total_votes', 'vine')
vine_df.show()
# Check for duplicates
vine_df.groupby(['review_id']).count().where('count > 1').sort('count', ascending=False).show()
# Check data types
vine_df.printSchema()

from pyspark.sql.types import IntegerType
# Change data type of column 'star_rating'
vine_df = vine_df.withColumn('star_rating', vine_df['star_rating'].cast(IntegerType()))
# Check data types
vine_df.printSchema()
vine_df.count()<jupyter_output><empty_output><jupyter_text>Load dataframes to the database<jupyter_code># Configure settings for RDS
mode = "append"
jdbc_url="jdbc:postgresql://database-assignment.cmf64jlceuwo.us-west-1.rds.amazonaws.com:5432/bd_assignment_db"
config = {"user":"postgres", 
          "password": "<PASSWORD>", 
          "driver":"org.postgresql.Driver"}
 # Write DataFrame to active_user table in 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="vine_table", mode=mode, properties=config)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.15647371650644304, 'fraction_numerical': 0.031294743301288605, 'mean_word_length': 6.298507462686567, 'pattern_counts': {'":': 3, '<': 16, '<?xml version=': 0, '>': 20, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '528706', 'license': 'ADSL', 'n_tokens_mistral': 1853, 'n_tokens_neox': 1735, 'n_words': 310, 'path': 'level-1/electronic_sport_etl.ipynb', 'repo_name': 'KeSavanh/big_data_challenge', 'size': '4892'}