<jupyter_start><jupyter_text>Big Data Project  **Participants:**  <NAME>(Cullen) **Date:** 05-10-2021 **Steps:**1. **Download Dataset**2. **Install the Scala Kernel And Set Up the Envoirment**3. **Install dependentcies**4. **Describe Data**5. **Data Preparation**6. **Algorithm Implementation** **Research Hypothesis:** * From my perception, customer review comments from Amazon have a high correlation with star ranking from customers.  With this perception, I made a few training model to predict the star rankning( which as the satifaction to the product) for the customers. **Research Questions:** 1. Does the frequency of words from comments affect the percentage of satisfaction?2. Is there a correlation between the percentage of the frequency with every word in comments?--- **Data Source:**1. **Amazon Review Data (2018)** : http://deepyeti.ucsd.edu/jianmo/amazon/2. **Affective Text** : http://web.eecs.umich.edu/~mihalcea/affectivetext  Step 1: Download Dataset<jupyter_code>!wget http://deepyeti.ucsd.edu/jianmo/amazon/categoryFilesSmall/Video_Games_5.json.gz
!gzip -d Video_Games_5.json.gz
!wget http://web.eecs.umich.edu/~mihalcea/affectivetext/data/WordNetAffectEmotionLists.tar.gz
!tar -xf WordNetAffectEmotionLists.tar.gz<jupyter_output>--2021-05-09 00:02:45--  http://deepyeti.ucsd.edu/jianmo/amazon/categoryFilesSmall/Video_Games_5.json.gz
Resolving deepyeti.ucsd.edu (deepyeti.ucsd.edu)... 192.168.127.12
Connecting to deepyeti.ucsd.edu (deepyeti.ucsd.edu)|192.168.127.12|:80... connected.
HTTP request sent, awaiting response... 200 OK
Length: 154050105 (147M) [application/octet-stream]
Saving to: ‘Video_Games_5.json.gz’

Video_Games_5.json. 100%[===================>] 146.91M  45.9MB/s    in 3.2s    

2021-05-09 00:02:48 (45.9 MB/s) - ‘Video_Games_5.json.gz’ saved [154050105/154050105]

--2021-05-09 00:02:53--  http://web.eecs.umich.edu/~mihalcea/affectivetext/data/WordNetAffectEmotionLists.tar.gz
Resolving web.eecs.umich.edu (web.eecs.umich.edu)... 172.16.58.3
Connecting to web.eecs.umich.edu (web.eecs.umich.edu)|172.16.58.3|:80... connected.
HTTP request sent, awaiting response... 200 OK
Length: 8287 (8.1K) [application/x-gzip]
Saving to: ‘WordNetAffectEmotionLists.tar.gz’

WordNetAffectEmotio 100%[=============[...]<jupyter_text>**Step 2:Install the Scala Kernel And Set Up the Envoirment**If you get a "scala" kernel not recognized warning when loading up the notebook for the first time, start by running the two cells below. Once you are done **reload the page** to load the notebook in the installed Scala kernel.<jupyter_code>!curl -Lo coursier https://git.io/coursier-cli && chmod +x coursier
!./coursier launch --fork almond:0.10.9 --scala 2.12 -- --install &> /dev/null 
println("(Testing)Hello, world!")<jupyter_output>(Testing)Hello, world!<jupyter_text>**Step 3: Install dependentcies**<jupyter_code>import $ivy.`org.slf4j:slf4j-api:2.0.0-alpha1`
import $ivy.`org.slf4j:slf4j-log4j12:2.0.0-alpha1`
import $ivy.`org.apache.spark::spark-core:3.1.1`
import $ivy.`org.apache.spark::spark-sql:3.1.1`
import $ivy.`org.apache.spark::spark-mllib:3.1.1`
// import $ivy.`com.nvidia::rapids-4-spark:0.4.1`
// import $ivy.`ai.rapids:cudf:0.18.1`
import $ivy.`log4j:log4j:1.2.17`
import $ivy.`org.jsoup:jsoup:1.13.1`
import $ivy.`org.plotly-scala::plotly-almond:0.8.2`
import coursierapi._
import sys.process._
import plotly._, plotly.element._, plotly.layout._, plotly.Almond._
import org.apache.log4j.{Level, Logger}
Logger.getLogger("org").setLevel(Level.OFF)
import org.apache.spark.sql._

implicit class RichDF(val df: DataFrame) {
  def showHTML(limit:Int = 20, truncate: Int = 20) = {
    import xml.Utility.escape
    val data = df.take(limit)
    val header = df.schema.fieldNames.toSeq
    val rows: Seq[Seq[String]] = data.map(_.toSeq.map(cell =>
        (cell match {
          case null => "null"
          case binary: Array[Byte] => binary.map("%02X".format(_)).mkString("[", " ", "]")
          case array: Array[_] => array.mkString("[", ", ", "]")
          case seq: Seq[_] => seq.mkString("[", ", ", "]")
          case _ => cell.toString
        }).take(truncate)
      ).map(p=> if(p.length >= truncate) p + "..." else p)
    )

    publish.html(s"""
      <table class="table dataframe" border="1">
        <tr>
        ${header.map(h => s"<th>${escape(h)}</th>").mkString}
        </tr>
        ${rows.map { row =>
          s"<tr>${row.map { c => s"<td>${escape(c)}</td>" }.mkString}</tr>"
        }.mkString
        }
      </table>""")
  }
}
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

val spark = NotebookSparkSession.builder()
    .appName("Product Satisfaction")
//     .config("spark.driver.resource.gpu.amount", "1")
    .config("spark.executor.memory","32g")
//     .config("spark.eventLog.enabled", "true")
    .config("spark.driver.cores", "4")
    .master("local[*]")
    .getOrCreate()
val sc = spark.sparkContext
val sqlContext= spark.sqlContext
// val conf = new SparkConf()
//             .set("spark.plugins", "com.nvidia.spark.SQLPlugin")
//             .set("spark.executor.resource.gpu.discoveryScript", "getGpusResources.sh")
//             .set("spark.rapids.sql.incompatibleOps.enabled", "true")
//             .set("spark.executor.resource.gpu.amount", "1")
//             .set("hadoop.home.dir","D:\\hadoop-3.2.2")
//             .set("spark.ui.enabled","true")
//             .setMaster("local[*]")
//             .setAppName("Product Satisfaction")


// val sc = new SparkContext(conf)
// val sqlContext= new SQLContext(sc)
// val spark=sqlContext.sparkSession

import spark.implicits._

//import org.jsoup.Jsoup
//kernel.publish.html(Jsoup.connect("http://8a1f1a0b9848:4040/").get().outerHtml())<jupyter_output>Loading spark-stubs
Getting spark JARs
Creating SparkSession<jupyter_text>**Step 4: Describe Data**<jupyter_code>val amazonComment = spark.read.json("res\\Video_Games_5.json")
amazonComment.printSchema()
amazonComment.showHTML()
amazonComment.count()

val hist = amazonComment.groupBy("overall").count()
val name = hist.select("overall").collect().map(_.getDouble(0)).toSeq
val value = hist.select("count").collect().map(_.getLong(0)).toSeq
val data = Seq(Bar(name,value))
plot(data)
import java.io.File
val files = new File("res\\WordNetAffectEmotionLists").listFiles.map(_.getName).filter(_!="README")
val emotionLables = (files.map(f => sc.textFile("WordNetAffectEmotionLists/"+ f ).map(_.split(' ').drop(1)).flatMap(l=>l).map(f.dropRight(4) -> _)).fold(sc.emptyRDD[(String, String)])(_ ++ _)).toDF("emotion","word")
emotionLables.showHTML()
import plotly.element._
val hist = emotionLables.groupBy("emotion").count()
val name = hist.select("emotion").collect().map(_.getString(0)).toSeq
val value = hist.select("count").collect().map(_.getLong(0)).toSeq
val data = Seq(Bar(name,value))
plot(data)<jupyter_output><empty_output><jupyter_text>**4.1 Transformate Data**<jupyter_code>import org.apache.spark.sql.functions.concat
val transformatedDF = amazonComment.select(($"overall" / 5).as("satisfaction") ,concat($"reviewText",$"summary").as("sentence"))
transformatedDF.showHTML()
transformatedDF.count()
transformatedDF.describe("satisfaction", "sentence").showHTML()<jupyter_output><empty_output><jupyter_text>**Step 5: Data Preparation**  5.1 Null Value Check<jupyter_code>val nullCheckedDF = transformatedDF.filter($"sentence".isNotNull)<jupyter_output><empty_output><jupyter_text>5.3 Sentence Separation<jupyter_code>import org.apache.spark.ml.feature.Tokenizer
val tokenizer = new Tokenizer().setInputCol("sentence").setOutputCol("words")
val wordsData = tokenizer.transform(nullCheckedDF.sample(0.5))
wordsData.showHTML()
wordsData.count()
wordsData.printSchema()

import scala.collection.mutable.WrappedArray
val emotionLablesMap = emotionLables.flatMap(p => Map(p.getString(1)-> p.getString(0))).collect().toMap
val emotionExtraction = wordsData.map(_.getAs[WrappedArray[String]](2).map(p => emotionLablesMap.getOrElse(p,null)).filter(_ !=null))
emotionExtraction.show()
val emotionExtractionHist = emotionExtraction.map(_.groupBy(l=>l).map(t =>(t._1-> t._2.size)).toSeq)
emotionExtractionHist.take(5)
emotionExtractionHist.show()<jupyter_output><empty_output><jupyter_text>5.3 Word2Vec Feature Transformation Model<jupyter_code>//Word2Vec
import org.apache.spark.ml.feature.Word2Vec
import org.apache.spark.ml.linalg.Vector
import org.apache.spark.sql.Row

// Input data: Each row is a bag of words from a sentence or document.


// Learn a mapping from words to Vectors.
val word2Vec = new Word2Vec()
  .setInputCol("words")
  .setOutputCol("result")
  .setVectorSize(128)
  .setMinCount(0)
val model = word2Vec.fit(wordsData)

val result = model.transform(wordsData)
result.showHTML()
result.describe().showHTML()
result.select("result").map(_.getAs[Vector])<jupyter_output><empty_output><jupyter_text>5.3 Seperate Training Dataset and Testing Dataset<jupyter_code>import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.sql.functions._
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.linalg.Vectors.fromML

val assembler = new VectorAssembler()
  .setInputCols(Array("result"))
  .setOutputCol("features")

val output = assembler.transform(result.select("satisfaction","result"))


val LabeledDF = output.map(p=>LabeledPoint(p.getDouble(0),fromML(p.getAs[Vector](1))))


val training,testing = LabeledDF.rdd.randomSplit(Array(0.7,0.3))
// val training_2,testing_2 = sc.union(testing).randomSplit(Array(0.7,0.3))

val sepTrainingDF = sc.union(training)
val sepTestingDF = sc.union(testing)

//StandardScaler
import org.apache.spark.mllib.feature.{StandardScaler, StandardScalerModel}

val trainScaler = new StandardScaler().fit(sepTrainingDF.map(p => p.features))
val trainScaleredData = sepTrainingDF.map(x=> (x.label, trainScaler.transform(x.features)))
val testScaler = new StandardScaler().fit(sepTestingDF.map(p => p.features))
val testScaleredData = sepTestingDF.map(x=> (x.label, testScaler.transform(x.features)))
import org.apache.spark.mllib.util.MLUtils
val trainingDF = MLUtils.convertVectorColumnsToML(trainScaleredData.toDF("label","features"))
val testingDF = MLUtils.convertVectorColumnsToML(testScaleredData.toDF("label","features"))
val labeledTrainingDF = trainingDF.map(p=>LabeledPoint(p.getDouble(0),fromML(p.getAs[Vector](1))))
val labeledTestingDF = testingDF.map(p=>LabeledPoint(p.getDouble(0),fromML(p.getAs[Vector](1))))<jupyter_output><empty_output><jupyter_text>**Step 7: Algorithm Implementation**  Generalized Linear regression---  Train<jupyter_code>import org.apache.spark.ml.regression.GeneralizedLinearRegression

val glr = new GeneralizedLinearRegression()
  .setFamily("gaussian")
  .setLink("identity")
  .setMaxIter(10)
  .setRegParam(0.3)

val glrModel = glr.fit(trainingDF)<jupyter_output><empty_output><jupyter_text>Evaluate<jupyter_code>// Print the coefficients and intercept for generalized linear regression model
println(s"Coefficients: ${glrModel.coefficients}")

println(s"Intercept: ${glrModel.intercept}")

// Summarize the model over the training set and print out some metrics
val summary = glrModel.summary
println(s"Coefficient Standard Errors: ${summary.coefficientStandardErrors.mkString(",")}")

println(s"T Values: ${summary.tValues.mkString(",")}")
println(s"P Values: ${summary.pValues.mkString(",")}")
println(s"Dispersion: ${summary.dispersion}")
println(s"Null Deviance: ${summary.nullDeviance}")
println(s"Residual Degree Of Freedom Null: ${summary.residualDegreeOfFreedomNull}")
println(s"Deviance: ${summary.deviance}")
println(s"Residual Degree Of Freedom: ${summary.residualDegreeOfFreedom}")
println(s"AIC: ${summary.aic}")
println("Deviance Residuals: ")
summary.residuals().showHTML()<jupyter_output>T Values: 9.387731152707884,17.633853133500832,1.434564687905509,-2.8431199814278627,2.753902049704036,-0.10391032030948247,3.0638764357711485,-11.737331203055438,5.264618101465167,-10.35288862608892,-10.436679533459362,-17.418869705359725,0.03255332255449339,8.013473158243022,8.208192601763947,-5.240490825350199,-14.558768884066483,30.558767877250048,12.210517127245344,-10.082817713142207,18.124960819667894,-11.631461987376657,6.0971860880339905,21.783408558795283,-4.573006595635398,5.078790541857327,9.634470922897485,-12.274101262676094,10.155722433618637,3.4730753058217076,-16.733095567264918,28.512987997616374,-5.65852437699466,16.466149035990664,-2.1180112890254286,-2.303790986119727,-5.269236653472048,-5.78933834389945,-2.019046479794775,-2.7538406978706185,13.109778964730335,-8.470667397408354,-7.992848548896884,-23.036984756266715,-10.204516145272718,8.750993866568098,2.1401226953937535,18.442269167198635,9.725256696796984,-1.8020241870573865,-8.48819349986203,16.81282119288876[...]<jupyter_text>Prediction Showcase<jupyter_code>val sample = testingDF.sample(0.1).take(10)
sample.map(p=>(p.getDouble(0),glrModel.predict(p.getAs[Vector](1))))<jupyter_output><empty_output><jupyter_text>Decision Trees Regression---  Train<jupyter_code>import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.feature.VectorIndexer
import org.apache.spark.ml.regression.DecisionTreeRegressionModel
import org.apache.spark.ml.regression.DecisionTreeRegressor

val featureIndexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexedFeatures")
  .setMaxCategories(4)
  .fit(trainingDF)

val dt = new DecisionTreeRegressor()
  .setLabelCol("label")
  .setFeaturesCol("indexedFeatures")

val pipeline = new Pipeline()
  .setStages(Array(featureIndexer, dt))

val dtModel = pipeline.fit(trainingDF)<jupyter_output><empty_output><jupyter_text>Evaluate<jupyter_code>val dtPredictions = dtModel.transform(testingDF)
// Select (prediction, true label) and compute test error.
val evaluator = new RegressionEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("rmse")
val rmse = evaluator.evaluate(dtPredictions)
println(s"Root Mean Squared Error (RMSE) on test data = $rmse")

val treeModel = dtModel.stages(1).asInstanceOf[DecisionTreeRegressionModel]
println(s"Learned regression tree model:\n ${treeModel.toDebugString}")<jupyter_output><empty_output><jupyter_text>Prediction Showcase<jupyter_code>// Make predictions.
dtPredictions.select("label","prediction").showHTML()
// sample.map(p=>(p.getDouble(0),model.predict(fromML(p.getAs[Vector](1)))))<jupyter_output><empty_output><jupyter_text>K-Means Clustering---  Train<jupyter_code>import org.apache.spark.ml.clustering.KMeans
import org.apache.spark.ml.evaluation.ClusteringEvaluator

// Trains a k-means model.
val kmeans = new KMeans().setK(10).setSeed(1L)
val kModel = kmeans.fit(trainingDF)


// Save and load model
//clusters.save(sc, "target/org/apache/spark/KMeansExample/KMeansModel")
//val sameModel = KMeansModel.load(sc, "target/org/apache/spark/KMeansExample/KMeansModel")<jupyter_output><empty_output><jupyter_text>Evaluate<jupyter_code>// Make predictions
val kPredictions = kModel.transform(testingDF)

// Evaluate clustering by computing Silhouette score
val evaluator = new ClusteringEvaluator()

val silhouette = evaluator.evaluate(kPredictions)
println(s"Silhouette with squared euclidean distance = $silhouette")
println("Test Error = " + (1.0 - silhouette))

// Shows the result.
println("Cluster Centers: ")
kModel.clusterCenters.foreach(println)<jupyter_output><empty_output><jupyter_text>Prediction Showcase<jupyter_code>kPredictions.select("label","prediction").showHTML()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1568904593639576, 'fraction_numerical': 0.07651782846129136, 'mean_word_length': 5.821209465381244, 'pattern_counts': {'":': 0, '<': 80, '<?xml version=': 0, '>': 111, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 1}, 'pii_count': 4, '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': '18', 'dir': 'jupyter-structured-clean-dedup', 'id': '332209', 'license': 'Apache-2.0', 'n_tokens_mistral': 6282, 'n_tokens_neox': 5529, 'n_words': 875, 'path': 'COMPSCI X409.2 Bigdata-Finail-Project/Big_Data_Final_Project_(scala).ipynb', 'repo_name': 'keyskull/Data-Science-Projects', 'size': '15574'}