本套技術專欄是作者（秦凱新）平時工作的總結和昇華，通過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和叢集環境容量規劃等內容，請持續關注本套部落格。版權宣告：禁止轉載，歡迎學習。QQ郵箱地址：[email protected]，如有任何商業交流，可隨時聯絡。

1 Spark ML流式線上學習初步講解

目前SparkStreaming 支援Streaming Linear Regression 和Streaming KMeans等。本文作為初步學習例項，以Streaming Linear Regression來說明問題，以方便後續更為複雜的學習案例。

2 初步案例實戰

import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.{LabeledPoint, StreamingLinearRegressionWithSGD}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import breeze.linalg.DenseVector

val traindir = sc.textFile("/data/train2.csv")
traindir.collect.take(2)

val ssc = new StreamingContext(sc,Seconds(10))
val stream=ssc.textFileStream("/data/train2.csv")
val NumFeatures = 11
val zeroVector = DenseVector.zeros[Double](NumFeatures)
val model = new StreamingLinearRegressionWithSGD().setInitialWeights(Vectors.dense(zeroVector.data)).setNumIterations(20).setRegParam(0.8).setStepSize(0.01)
  
  
val labeledStream = stream.map{ event =>
val split = event.split(",")
val y = split(11).toDouble
val features = split(1).slice(0,11).map(_.toDouble)
LabeledPoint(label = y, features = Vectors.dense(features))
}

model.trainOn(labeledStream)
model.predictOn(labeledStream.map(lp=>(lp.label,lp.features))).print
ssc.start()
ssc.awaitTermination()
  }
}
 

3 總結

線上學習仍然採用老的mlib包，可謂前景堪憂啊。

秦凱新 20181119