Spark机器学习

1 在线学习

模型随着接收的新消息，不断更新自己；而不是像离线训练一次次重新训练。

2 Spark Streaming

• 离散化流(DStream)
• 输入源：Akka actors、消息队列、Flume、Kafka、……
  

http://spark.apache.org/docs/latest/streaming-programming-guide.html

• 类群(lineage)：应用到RDD上的转换算子和执行算子的集合

3 MLib+Streaming应用

3.0 build.sbt

依赖Spark MLlib和Spark Streaming

name := "scala-spark-streaming-app"version := "1.0"scalaVersion := "2.11.7"libraryDependencies += "org.apache.spark" %% "spark-mllib" % "1.5.1"libraryDependencies += "org.apache.spark" %% "spark-streaming" % "1.5.1"

使用国内镜像仓库

~/.sbt/repositories

[repositories]
local
osc: http://maven.oschina.net/content/groups/public/
typesafe: http://repo.typesafe.com/typesafe/ivy-releases/, [organization]/[module]/(scala_[scalaVersion]/)(sbt_[sbtVersion]/)[revision]/[type]s/[artifact](-[classifier]).[ext], bootOnly
sonatype-oss-releases
maven-central
sonatype-oss-snapshots

3.1 生产消息

object StreamingProducer {def main(args: Array[String]) {val random = new Random()// Maximum number of events per secondval MaxEvents = 6// Read the list of possible namesval namesResource = this.getClass.getResourceAsStream("/names.csv")val names = scala.io.Source.fromInputStream(namesResource).getLines().toList.head.split(",").toSeq// Generate a sequence of possible productsval products = Seq("iPhone Cover" -> 9.99,"Headphones" -> 5.49,"Samsung Galaxy Cover" -> 8.95,"iPad Cover" -> 7.49)/** Generate a number of random product events */def generateProductEvents(n: Int) = {(1 to n).map { i =>val (product, price) = products(random.nextInt(products.size))val user = random.shuffle(names).head(user, product, price)}}// create a network producerval listener = new ServerSocket(9999)println("Listening on port: 9999")while (true) {val socket = listener.accept()new Thread() {override def run = {println("Got client connected from: " + socket.getInetAddress)val out = new PrintWriter(socket.getOutputStream(), true)while (true) {Thread.sleep(1000)val num = random.nextInt(MaxEvents)val productEvents = generateProductEvents(num)productEvents.foreach{ event =>out.write(event.productIterator.mkString(","))out.write("\n")}out.flush()println(s"Created $num events...")}socket.close()}}.start()}}
}

sbt runMultiple main classes detected, select one to run:[1] MonitoringStreamingModel[2] SimpleStreamingApp[3] SimpleStreamingModel[4] StreamingAnalyticsApp[5] StreamingModelProducer[6] StreamingProducer[7] StreamingStateAppEnter number: 6

3.2 打印消息

object SimpleStreamingApp {def main(args: Array[String]) {val ssc = new StreamingContext("local[2]", "First Streaming App", Seconds(10))val stream = ssc.socketTextStream("localhost", 9999)// here we simply print out the first few elements of each batchstream.print()ssc.start()ssc.awaitTermination()}
}

sbt runEnter number: 2

3.3 流式分析

object StreamingAnalyticsApp {def main(args: Array[String]) {val ssc = new StreamingContext("local[2]", "First Streaming App", Seconds(10))val stream = ssc.socketTextStream("localhost", 9999)// create stream of events from raw text elementsval events = stream.map { record =>val event = record.split(",")(event(0), event(1), event(2))}/*We compute and print out stats for each batch.Since each batch is an RDD, we call forEeachRDD on the DStream, and apply the usual RDD functionswe used in Chapter 1.*/events.foreachRDD { (rdd, time) =>val numPurchases = rdd.count()val uniqueUsers = rdd.map { case (user, _, _) => user }.distinct().count()val totalRevenue = rdd.map { case (_, _, price) => price.toDouble }.sum()val productsByPopularity = rdd.map { case (user, product, price) => (product, 1) }.reduceByKey(_ + _).collect().sortBy(-_._2)val mostPopular = productsByPopularity(0)val formatter = new SimpleDateFormatval dateStr = formatter.format(new Date(time.milliseconds))println(s"== Batch start time: $dateStr ==")println("Total purchases: " + numPurchases)println("Unique users: " + uniqueUsers)println("Total revenue: " + totalRevenue)println("Most popular product: %s with %d purchases".format(mostPopular._1, mostPopular._2))}// start the contextssc.start()ssc.awaitTermination()}
}

sbt runEnter number: 4

3.4 有状态的流计算

object StreamingStateApp {import org.apache.spark.streaming.StreamingContext._def updateState(prices: Seq[(String, Double)], currentTotal: Option[(Int, Double)]) = {val currentRevenue = prices.map(_._2).sumval currentNumberPurchases = prices.sizeval state = currentTotal.getOrElse((0, 0.0))Some((currentNumberPurchases + state._1, currentRevenue + state._2))}def main(args: Array[String]) {val ssc = new StreamingContext("local[2]", "First Streaming App", Seconds(10))// for stateful operations, we need to set a checkpoint locationssc.checkpoint("/tmp/sparkstreaming/")val stream = ssc.socketTextStream("localhost", 9999)// create stream of events from raw text elementsval events = stream.map { record =>val event = record.split(",")(event(0), event(1), event(2).toDouble)}val users = events.map { case (user, product, price) => (user, (product, price)) }val revenuePerUser = users.updateStateByKey(updateState)revenuePerUser.print()// start the contextssc.start()ssc.awaitTermination()}
}

sbt runEnter number: 7

4 线性流回归

线性回归StreamingLinearRegressionWithSGD

• trainOn
• predictOn

4.1 流数据生成器

object StreamingModelProducer {import breeze.linalg._def main(args: Array[String]) {// Maximum number of events per secondval MaxEvents = 100val NumFeatures = 100val random = new Random()/** Function to generate a normally distributed dense vector */def generateRandomArray(n: Int) = Array.tabulate(n)(_ => random.nextGaussian())// Generate a fixed random model weight vectorval w = new DenseVector(generateRandomArray(NumFeatures))val intercept = random.nextGaussian() * 10/** Generate a number of random product events */def generateNoisyData(n: Int) = {(1 to n).map { i =>val x = new DenseVector(generateRandomArray(NumFeatures))val y: Double = w.dot(x)val noisy = y + intercept //+ 0.1 * random.nextGaussian()(noisy, x)}}// create a network producerval listener = new ServerSocket(9999)println("Listening on port: 9999")while (true) {val socket = listener.accept()new Thread() {override def run = {println("Got client connected from: " + socket.getInetAddress)val out = new PrintWriter(socket.getOutputStream(), true)while (true) {Thread.sleep(1000)val num = random.nextInt(MaxEvents)val data = generateNoisyData(num)data.foreach { case (y, x) =>val xStr = x.data.mkString(",")val eventStr = s"$y\t$xStr"out.write(eventStr)out.write("\n")}out.flush()println(s"Created $num events...")}socket.close()}}.start()}}
}

sbt runEnter number: 5

4.2 流回归模型

object SimpleStreamingModel {def main(args: Array[String]) {val ssc = new StreamingContext("local[2]", "First Streaming App", Seconds(10))val stream = ssc.socketTextStream("localhost", 9999)val NumFeatures = 100val zeroVector = DenseVector.zeros[Double](NumFeatures)val model = new StreamingLinearRegressionWithSGD().setInitialWeights(Vectors.dense(zeroVector.data)).setNumIterations(1).setStepSize(0.01)// create a stream of labeled pointsval labeledStream: DStream[LabeledPoint] = stream.map { event =>val split = event.split("\t")val y = split(0).toDoubleval features: Array[Double] = split(1).split(",").map(_.toDouble)LabeledPoint(label = y, features = Vectors.dense(features))}// train and test model on the stream, and print predictions for illustrative purposesmodel.trainOn(labeledStream)//model.predictOn(labeledStream).print()ssc.start()ssc.awaitTermination()}
}

sbt runEnter number: 5

5 流K-均值

• K-均值聚类：StreamingKMeans

6 评估

object MonitoringStreamingModel {def main(args: Array[String]) {val ssc = new StreamingContext("local[2]", "First Streaming App", Seconds(10))val stream = ssc.socketTextStream("localhost", 9999)val NumFeatures = 100val zeroVector = DenseVector.zeros[Double](NumFeatures)val model1 = new StreamingLinearRegressionWithSGD().setInitialWeights(Vectors.dense(zeroVector.data)).setNumIterations(1).setStepSize(0.01)val model2 = new StreamingLinearRegressionWithSGD().setInitialWeights(Vectors.dense(zeroVector.data)).setNumIterations(1).setStepSize(1.0)// create a stream of labeled pointsval labeledStream = stream.map { event =>val split = event.split("\t")val y = split(0).toDoubleval features = split(1).split(",").map(_.toDouble)LabeledPoint(label = y, features = Vectors.dense(features))}// train both models on the same streammodel1.trainOn(labeledStream)model2.trainOn(labeledStream)// use transform to create a stream with model error ratesval predsAndTrue = labeledStream.transform { rdd =>val latest1 = model1.latestModel()val latest2 = model2.latestModel()rdd.map { point =>val pred1 = latest1.predict(point.features)val pred2 = latest2.predict(point.features)(pred1 - point.label, pred2 - point.label)}}// print out the MSE and RMSE metrics for each model per batchpredsAndTrue.foreachRDD { (rdd, time) =>val mse1 = rdd.map { case (err1, err2) => err1 * err1 }.mean()val rmse1 = math.sqrt(mse1)val mse2 = rdd.map { case (err1, err2) => err2 * err2 }.mean()val rmse2 = math.sqrt(mse2)println(s"""|-------------------------------------------|Time: $time|-------------------------------------------""".stripMargin)println(s"MSE current batch: Model 1: $mse1; Model 2: $mse2")println(s"RMSE current batch: Model 1: $rmse1; Model 2: $rmse2")println("...\n")}ssc.start()ssc.awaitTermination()}
}

sbt runEnter number: 1