//Scala
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.feature.HashingTF
import org.apache.spark.mllib.classification.LogisticRegressionWithSGD

//Scala
val spam = sc.textFile("/FileStore/tables/spam.txt")
val ham = sc.textFile("/FileStore/tables/ham.txt")

//Scala
val tf = new HashingTF(numFeatures = 230000)

//Scala
val spamFeatures = spam.map(email => tf.transform(email.split(" ")))
val hamFeatures = ham.map(email => tf.transform(email.split(" ")))

//Scala
val positiveExamples = spamFeatures.map(features => LabeledPoint(1, features))
val negativeExamples = hamFeatures.map(features => LabeledPoint(0, features))
val trainingData = positiveExamples ++ negativeExamples

//Scala
trainingData.cache()

//Scala
val lrLearner = new LogisticRegressionWithSGD()
val model = lrLearner.run(trainingData)

//Scala
val posTestExample = tf.transform("Oh my god, you're a winner! To take your price please send 3$ to this card".split(" "))
val negTestExample = tf.transform("Hi Dad, I started studying Spark the other ...".split(" "))
println(s"Prediction for positive test example: ${model.predict(posTestExample)}")
println(s"Prediction for negative test example: ${model.predict(negTestExample)}")