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

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

//Создаем экземпляр HashingTF для отображения текста электронных писем в векторы с 230000 признаков.
val tf = new HashingTF(numFeatures = 230000)

//Разбиваем каждое электронное письмо на слова и каждое слово отображаем в один признак.
val spamFeatures = spam.map(email => tf.transform(email.split(" ")))
val hamFeatures = ham.map(email => tf.transform(email.split(" ")))

//Создаем наборы данных LabeledPoint для примеров, дающих положительную реакцию (спам) и отрицательную (релевантные письма).
val positiveExamples = spamFeatures.map(features => LabeledPoint(1, features))
val negativeExamples = hamFeatures.map(features => LabeledPoint(0, features))
val trainingData = positiveExamples ++ negativeExamples

//Выполняем кэширование обучающей выборки, потому что алгоритм Logistic Regression является циклическим.
calatrainingData.cache()

//Проводим обучение модели с помощью логистической регрессии методом SGD.
val lrLearner = new LogisticRegressionWithSGD()
val model = lrLearner.run(trainingData)

//Проверяем положительный и отрицательный экземпляры. Сначала применяем то же преобразование HashingTF, чтобы получить векторы признаков, затем применяем модель.
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)}")