# ---
# 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
# ---

# + id="V58rxea0HqSa" colab={"base_uri": "https://localhost:8080/"} outputId="a9b62101-4f2e-4fc6-86ae-167b5754294d"
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.3'
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()

# + id="_xKwTpATHqSe" colab={"base_uri": "https://localhost:8080/"} outputId="5ab64a10-f498-438c-a341-ff5fc079e004"
# Download the Postgres driver that will allow Spark to interact with Postgres.
# !wget https://jdbc.postgresql.org/download/postgresql-42.2.16.jar

# + id="MMqDAjVS0KN9"
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("M16-Amazon-Challenge").config("spark.driver.extraClassPath","/content/postgresql-42.2.16.jar").getOrCreate()

# + [markdown] id="cyBsySGuY-9V"
# ### Load Amazon Data into Spark DataFrame

# + id="CtCmBhQJY-9Z" colab={"base_uri": "https://localhost:8080/"} outputId="15da4527-c3a2-451f-b3c3-28b1ad88b7f8"
from pyspark import SparkFiles
url = "https://s3.amazonaws.com/amazon-reviews-pds/tsv/amazon_reviews_us_Software_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()

# + [markdown] id="2yUSe55VY-9t"
# ### Create DataFrames to match tables

# + id="C8REmY1aY-9u"
from pyspark.sql.functions import to_date
# Read in the Review dataset as a DataFrame


# + id="B0TESUDRY-90" colab={"base_uri": "https://localhost:8080/"} outputId="28675d19-d8c9-44df-d6be-1f5cd5c5f1e9"
# Create the customers_table DataFrame
customers_df = df.groupby("customer_id").count().withColumnRenamed("count", "customer_count")
customers_df.show()


# + id="4FwXA6UvY-96" colab={"base_uri": "https://localhost:8080/"} outputId="4d349584-bede-424f-eef9-002960e7f5a3"
# Create the products_table DataFrame and drop duplicates. 
products_df = df.select(["product_id","product_title"]).drop_duplicates()
products_df.show()

# + id="MkqyCuNQY-9-" colab={"base_uri": "https://localhost:8080/"} outputId="d80103ca-3d17-4d9e-f1c1-a37b12359f44"
# 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()

# + id="lzMmkdKmY--D" colab={"base_uri": "https://localhost:8080/"} outputId="950a299f-56ed-434a-9def-93b27cb6e9d6"
# Create the vine_table. DataFrame
vine_df = df.select(["review_id","star_rating","helpful_votes","total_votes","vine","verified_purchase"])
vine_df.show()

# + id="7jiUvs1aY--L"
from pyspark.sql.functions import col

# + id="T2zgZ-aKY--Q" colab={"base_uri": "https://localhost:8080/"} outputId="6dc964a4-c7dc-4351-db27-6abb30c91825"
filtered_df = df.filter(col("total_votes") >=10 )
filtered_df.show()

# + id="1m3yzn-LY--U" colab={"base_uri": "https://localhost:8080/"} outputId="ada5da7c-7b95-4515-9bc1-6d184f0f34a8"
vote_ratio_df = filtered_df.filter(col("helpful_votes")/col("total_votes")>=0.5)
vote_ratio_df.show()

# + id="KbXri15fY--Z" colab={"base_uri": "https://localhost:8080/"} outputId="5e42636e-24b5-4302-ae98-327e58f656ad"
paid_review_df = vote_ratio_df.filter(col("vine") == "Y")
paid_review_df.show()

# + id="XdQknSHLY--e" colab={"base_uri": "https://localhost:8080/"} outputId="d937a98e-065c-48bb-e435-829c53b95d5f"
unpaid_review_df = vote_ratio_df.filter(col("vine") == "N")
unpaid_review_df.show()

# + id="Exuo6ebUsCqW" colab={"base_uri": "https://localhost:8080/"} outputId="49342143-40e6-4c41-ce3a-7b565a00f103"
# PAID
# total reviews:
total_paid_reviews = paid_review_df.count()
print(total_paid_reviews)



# + colab={"base_uri": "https://localhost:8080/"} id="q4a_4tyb6etb" outputId="8ded3b4a-de29-4665-eceb-1ed7cfe6e975"
# 5 star reviews:
paid_five_star_reviews = paid_review_df.filter(col("star_rating")==5).count()
print(paid_five_star_reviews)


# + colab={"base_uri": "https://localhost:8080/"} id="o3hDLlOd9btf" outputId="c8554863-b26c-4d62-bed3-79b2bc7c742e"
# % 5 star reviews
paid_ratio = paid_five_star_reviews/total_paid_reviews
print(paid_ratio)


# + colab={"base_uri": "https://localhost:8080/"} id="A7ahCnlr9dHF" outputId="bf102078-50f3-4bdf-a6ca-6bcc8b52435f"
# UNPAID
# total reviews:
total_unpaid_reviews = unpaid_review_df.count()
print(total_unpaid_reviews)



# + colab={"base_uri": "https://localhost:8080/"} id="2JS_ahhB9ev9" outputId="7864a4b5-64ac-443b-ae83-5196ab68c705"
# 5 star reviews:
unpaid_five_star_reviews = unpaid_review_df.filter(col("star_rating")==5).count()
print(unpaid_five_star_reviews)


# + id="rwPtdrEc9gEd"
# % 5 star reviews
unpaid_ratio = unpaid_five_star_reviews/total_unpaid_reviews
print(unpaid_ratio)

StarcoderdataJupyterScriptsDedupFiltered