<jupyter_start><jupyter_text>Analyzing Amazon product review data set  using dataframe<jupyter_code>import os
import sys
# os.environ["PYSPARK_PYTHON"] = "/opt/cloudera/parcels/Anaconda/bin/python"
# os.environ["JAVA_HOME"] = "/usr/java/jdk1.8.0_161/jre"
# os.environ["SPARK_HOME"] = "/opt/cloudera/parcels/SPARK2-2.3.0.cloudera2-1.cdh5.13.3.p0.316101/lib/spark2/"
os.environ["PYLIB"] = os.environ["SPARK_HOME"] + r"\python\lib"
sys.path.insert(0, os.environ["PYLIB"] +r"\py4j-0.10.9-src.zip")
sys.path.insert(0, os.environ["PYLIB"] +r"\pyspark.zip")
from pyspark.sql.functions import *

#  Customs User defined functions (UDFs) need to be register
from pyspark.sql.types import *
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("AmazonReviewDemo").getOrCreate()
spark
# Reading data from S3 to a DataFrame
# reviews_df = spark.read.json('s3a://sparkbucket123/Electronics_5.json')
reviews_df = spark.read.json(r'F:\Users\SunilYadav\Desktop\Electronics_5.json')
reviews_df.show(5)
reviews_df.printSchema()
# Finding the number of individual reviewers
distinct_reviewers = reviews_df.select("reviewerID").distinct()
distinct_reviewers.count()
# Finding the number of product counts
reviews_df.select("asin").distinct().count()

# Find the products with reviews
reviews_df.select("asin").groupBy("asin").count().orderBy("count" , ascending =False)
# Find the number of reviews given by each user
reviews_df.select("reviewerID").groupBy("reviewerID").count().orderBy("count" , ascending =False).show()
# Number of user reviews every month sorted in decreasing order of reviews count

from pyspark.sql.functions import *
reviews_df1= reviews_df.withColumn("ts" , reviews_df['unixReviewTime'].cast('timestamp'))
reviews_df1.printSchema()

reviews_df_with_year= reviews_df1.withColumn("year" , year("ts")).withColumn("month" , month("ts"))
reviews_df_with_year.groupBy("year", "month").count().orderBy('count', ascending = False).show()
# Find how many times each prodcut has been reviewed great
great_reviews = reviews_df.filter(reviews_df.reviewText.contains("great"))
great_reviews.groupBy("asin").count().orderBy('count', ascending = False).show()
lst = list(range(100))
sc = spark.sparkContext
rdd_lst = sc.parallelize(lst)
type(rdd_lst)
rdd_lst.collect()
lst_df = rdd_lst.toDF("name:int")
type(lst_df)
lst_df.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.150856665273715, 'fraction_numerical': 0.01880484747179273, 'mean_word_length': 6.298780487804878, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '628439', 'license': 'Apache-2.0', 'n_tokens_mistral': 889, 'n_tokens_neox': 860, 'n_words': 152, 'path': 'zExtraLearning/Amazon+Dataset.ipynb', 'repo_name': 'talk2sunil83/UpgradLearning', 'size': '2393'}