# SPARK Using Python and Scala

import org.apache.spark.sql.Encoders

case class flight(
    DayofMonth:Int, 
    DayOfWeek:Int, 
    Carrier:String, 
    OriginAirportID:Int, 
    DestAirportID:Int, 
    DepDelay:Int, 
    ArrDelay:Int)

val flightSchema = Encoders.product[flight].schema

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

val data = spark.createDataFrame(Seq(
 ("A", 1),
 ("B", 2),
 ("C", 3),
 ("Very long text to show", 4))).toDF("text", "value")

data.show()

val flights = spark.read.schema(flightSchema).
option("header", "true").
csv("wasb:///data/raw-flight-data.csv")

airports = spark.read.csv('wasb:///data/airports.csv', 
					header=True, 
					inferSchema=True)

airports.printSchema()

flights.show()

flights.show(30)

data.show(5,truncate = false)

flights.count()

flights.describe().show()

val cities = airports.select("city", "name")
cities.show()

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

flights.select( $"ArrDelay", 
               ($"ArrDelay" > 15).cast("Int").alias("Late")
              )

flights.filter("ArrDelay==24")

flights.dropDuplicates()

flights.na.drop("any", Array("ArrDelay", "DepDelay"))

flights.na.fill(0, Array("ArrDelay", "DepDelay"))

val flightsByOrigin = flights.join(airports, 
                                   $"OriginAirportID" === $"airport_id"
                                  ).groupBy("city").count()
flightsByOrigin.show()

flights.stat.corr("DepDelay","ArrDelay")

val data = flights.withColumnRenamed("ArrDelay", "ArrDelayNewLabel")

flights.createOrReplaceTempView("flightData")
spark.sql("SELECT DayOfWeek, AVG(ArrDelay) AS AvgDelay FROM flightData GROUP BY DayOfWeek ORDER BY DayOfWeek").show()

val splits = flights.randomSplit(Array(0.7, 0.3))
val train = splits(0)
val test = splits(1)

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature.{VectorAssembler, StringIndexer, VectorIndexer, MinMaxScaler}

import org.apache.spark.ml.feature.StringIndexer

val strIdx = new StringIndexer().setInputCol("Carrier").setOutputCol("CarrierIdx")

val data2 = strIdx.fit(train).transform(train)
data2.select("Carrier","carrierIdx").show(5)

val assembler = new VectorAssembler().setInputCols(
    Array("DayofMonth", "DayOfWeek", "OriginAirportID")).
    setOutputCol("features")

val training = assembler.transform(train)
training.select("DayofMonth", 
                "DayOfWeek", 
                "OriginAirportID",
                "features").show(5)

import org.apache.spark.ml.feature.VectorIndexer

val catVect = new VectorAssembler().setInputCols(Array("CarrierIdx", "DayofMonth", "DayOfWeek", "OriginAirportID", "DestAirportID")).setOutputCol("catFeatures")
val data3 =  catVect.transform(data2)


val catIdx = new VectorIndexer().
        setInputCol(catVect.getOutputCol).
        setOutputCol("idxCatFeatures")



val data4 = catIdx.fit(data3).transform(data3)
data4.select("catFeatures","idxCatFeatures").show(5)

import org.apache.spark.ml.feature.MinMaxScaler

val numVect = new VectorAssembler().
	setInputCols(Array("DepDelay")).
	setOutputCol("numFeatures")
val data5  = numVect.transform(data4)


val minMax = new MinMaxScaler().
    setInputCol(numVect.getOutputCol).
    setOutputCol("normFeatures")


val data6 = minMax.fit(data5).transform(data5)
data6.select("numFeatures","normFeatures").show(5)

import org.apache.spark.ml.Pipeline

import org.apache.spark.ml.classification.DecisionTreeClassifier
val featVect = new VectorAssembler().
setInputCols(Array("idxCatFeatures", "normFeatures")).
setOutputCol("features")

val dt = new DecisionTreeClassifier().
setLabelCol("label").
setFeaturesCol("features")

val pipeline = new Pipeline().
setStages(Array(strIdx, catVect, catIdx, 
                numVect, minMax, featVect, dt))

val model = pipeline.fit(train)
model.transform(test).select(
    "trueLabel", "probability","prediction").show(15)

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.LogisticRegression

val lr = new LogisticRegression().
            setLabelCol("label").
            setFeaturesCol("features").
            setMaxIter(10).
            setRegParam(0.3).
            setThreshold(0.35)
val pipeline = new Pipeline().setStages(Array(assembler, lr))

// Train the model
val model = pipeline.fit(train)

val predicted = model.transform(test).
            select("features", "label","probability", "prediction")

import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator


val evaluator = new BinaryClassificationEvaluator().
        setLabelCol("label").
        setRawPredictionCol("prediction").
        setMetricName("areaUnderROC")

val auc = evaluator.evaluate(predicted)
println(auc)

import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator

val evaluator = new MulticlassClassificationEvaluator().
        setLabelCol("label").
        setPredictionCol("prediction").
        setMetricName("accuracy")

val auc = evaluator.evaluate(predicted)
println(auc)

val tp = predicted.filter("prediction == 1 AND label == 1").count().toFloat
val fp = predicted.filter("prediction == 1 AND label == 0").count().toFloat
val tn = predicted.filter("prediction == 0 AND label == 0").count().toFloat
val fn = predicted.filter("prediction == 0 AND label == 1").count().toFloat
val metrics = spark.createDataFrame(Seq(
 ("TP", tp),
 ("FP", fp),
 ("TN", tn),
 ("FN", fn),
 ("Precision", tp / (tp + fp)),
 ("Recall", tp / (tp + fn)))).toDF("metric", "value")
metrics.show()

import org.apache.spark.ml.regression.LinearRegression

val lr = new LinearRegression().
	setLabelCol("label").
	setFeaturesCol("features").
	setMaxIter(10).
	setRegParam(0.3)

val pipeline = new Pipeline().setStages(Array(assembler, lr))

// Train the model
val model = pipeline.fit(train)

val prediction = model.transform(test)
prediction.select("features", "prediction", "label").show()

import org.apache.spark.ml.evaluation.RegressionEvaluator

val evaluator = new RegressionEvaluator().
setLabelCol("label").
setPredictionCol("prediction").
setMetricName("rmse")

val rmse = evaluator.evaluate(prediction)
println("Root Mean Square Error (RMSE): " + (rmse))

import org.apache.spark.ml.tuning.{ParamGridBuilder, TrainValidationSplit}

val paramGrid = new ParamGridBuilder().
	addGrid(lr.regParam, Array(0.3, 0.1, 0.01)).
	addGrid(lr.maxIter, Array(10, 5)).
	addGrid(lr.threshold, Array(0.35, 0.3)).
	build()

val tvs = new TrainValidationSplit().
	setEstimator(pipeline).
	setEvaluator(new BinaryClassificationEvaluator).
	setEstimatorParamMaps(paramGrid).
	setTrainRatio(0.8)

val model = tvs.fit(train)

val prediction = model.transform(test)

val tp = prediction.filter("prediction == 1 AND truelabel == 1").count().toFloat
val fp = prediction.filter("prediction == 1 AND truelabel == 0").count().toFloat
val tn = prediction.filter("prediction == 0 AND truelabel == 0").count().toFloat
val fn = prediction.filter("prediction == 0 AND truelabel == 1").count().toFloat
val metrics = spark.createDataFrame(Seq(
 ("TP", tp),
 ("FP", fp),
 ("TN", tn),
 ("FN", fn),
 ("Precision", tp / (tp + fp)),
 ("Recall", tp / (tp + fn)))).toDF("metric", "value")
metrics.show()

val evaluator = new BinaryClassificationEvaluator().
	setLabelCol("trueLabel").
	setRawPredictionCol("prediction").
	setMetricName("areaUnderROC")
val aur = evaluator.evaluate(prediction)
println("AUR = " + (aur))

import org.apache.spark.ml.tuning.{ParamGridBuilder, CrossValidator}

val paramGrid = new ParamGridBuilder().
    addGrid(lr.regParam, Array(0.3, 0.01)).
    addGrid(lr.maxIter, Array(10, 5)).
    build()
val cv = new CrossValidator().
    setEstimator(pipeline).
    setEvaluator(new RegressionEvaluator).
    setEstimatorParamMaps(paramGrid).setNumFolds(2)

val model = cv.fit(train)

import org.apache.spark.ml.feature.Tokenizer


val tokenizer = new Tokenizer().
    setInputCol("SentimentText").
    setOutputCol("SentimentWords")

val tokenized = tokenizer.transform(train)
tokenized.select("SentimentText",
                 "SentimentWords").show(5,truncate = false)

import org.apache.spark.ml.feature.StopWordsRemover

val swr = new StopWordsRemover().
    setInputCol(tokenizer.getOutputCol).
    setOutputCol("MeaningfulWords")

val swr_output = swr.transform(tokenized)
swr_output.select(tokenizer.getOutputCol,
                  "MeaningfulWords").show(5, truncate = false)

import org.apache.spark.ml.feature.HashingTF

val hashTF = new HashingTF().
    setInputCol(swr.getOutputCol). 
    setOutputCol("features")

val hashTF_output = hashTF.transform(swr_output)
hashTF_output.select(
    swr.getOutputCol,
    "features").show(5, truncate = false)

val lr = new LogisticRegression().
    setLabelCol("label").
    setFeaturesCol("features").
    setMaxIter(10).
    setRegParam(0.01)

val pipeline = new Pipeline().
    setStages(Array(tokenizer, swr, hashTF, lr))

val piplineModel = pipeline.fit(train)
val prediction = piplineModel.transform(test)
prediction.select("SentimentText", 
                  "features", 
                  "prediction", 
                  "trueLabel").show(10,truncate = false)

import org.apache.spark.ml.recommendation.ALS

val als = new ALS().
    setMaxIter(5).
    setRegParam(0.01).
    setUserCol("userId").
    setItemCol("movieId").
    setRatingCol("label")

val model = als.fit(train)
val prediction = model.transform(test)

prediction.join(movies, "movieId").
    select("userId", "title", "prediction", "trueLabel").
    show(100, truncate=false)

import org.apache.spark.ml.clustering.KMeans

val kmeans = new KMeans().
    setFeaturesCol(assembler.getOutputCol).
    setPredictionCol("cluster").
    setK(5).
    setSeed(0)

val model = kmeans.fit(train)

model.clusterCenters

val prediction = model.transform(train)

prediction.groupBy("cluster").
	count().
	orderBy("cluster").
	show()
prediction.select("CustomerName", 
                  "cluster").show(50)