22
правки
Изменения
→Примеры реализации алгоритмов с использованием Spark MLlib
== Примеры реализации алгоритмов с использованием Spark MLlib ==
Рассмотрим удобство использования Apache Spark на примере. Решим задачу “Hello world” из мира Big Data - подсчет слов в файле.
sparkContext.textFile("hdfs://...")
.flatMap(line => line.split(" "))
.map(word => (word, 1)).reduceByKey(_ + _)
.saveAsTextFile("hdfs://...")
Рассмотрим еще один пример анализа данных с помощью Spark MLlib. Поставим задачу проанализировать наличие землетрясений на основе твитов. Отберем необходимые твиты с информацией о “тряске” или “землетрясении”.
TwitterUtils.createStream(...)
.filter(_.getText.contains("earthquake") || _.getText.contains("shaking"))
Далее подготовим эти данные к построению модели. Разобьем выборку на обучающую и тестирующую
val data = MLUtils.loadLibSVMFile(sc, "sample_earthquate_tweets.txt")
val splits = data.randomSplit(Array(0.6, 0.4), seed = 11L)
val training = splits(0).cache()
val test = splits(1)
Проводим обучение нашей модели. В качестве алгоритма выберем метод опорных векторов.
val numIterations = 100
val model = SVMWithSGD.train(training, numIterations)
// Очищаем пороговое значение, заданное по умолчанию
model.clearThreshold()
Выведем интересующий нас результат
val metrics = new BinaryClassificationMetrics(scoreAndLabels)
val auROC = metrics.areaUnderROC()
println("Area under ROC = " + auROC)
Если результат нас устраивает можем дополнить наш код отправкой сообщений о землетрясении пользователям с помощью Spark SQL и пользовательской функции sendEMail()
val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)
sqlContext.sql("FROM earthquake_warning_users SELECT firstName, lastName, city, email")
.collect().foreach(sendEmail())
Еще одной интересной особенностью Spark MLlib является возможность кастомизировать выбранные алгоритмы. К примеру, можно увеличить число итераций или изменить параметр регуляризации
import org.apache.spark.mllib.optimization.L1Updater
val svmAlg = new SVMWithSGD()
svmAlg.optimizer.
setNumIterations(200).
setRegParam(0.1).
setUpdater(new L1Updater)
val modelL1 = svmAlg.run(training)
== См. также ==
