spark機(jī)器學(xué)習(xí)-聚類

聚類算法是一種無監(jiān)督學(xué)習(xí)任務(wù)，用于將對(duì)象分到具有高度相似性的聚類中，聚類算法的思想簡(jiǎn)單的說就是物以類聚的思想，相同性質(zhì)的點(diǎn)在空間中表現(xiàn)的較為緊密和接近，主要用于數(shù)據(jù)探索與異常檢測(cè)，最常用的一種聚類算法是K均值(K-means)聚類算法

算法原理
kmeans的計(jì)算方法如下：
1 選取k個(gè)中心點(diǎn)
2 遍歷所有數(shù)據(jù)，將每個(gè)數(shù)據(jù)劃分到最近的中心點(diǎn)中
3 計(jì)算每個(gè)聚類的平均值，并作為新的中心點(diǎn)
4 重復(fù)2-3，直到這k個(gè)中線點(diǎn)不再變化（收斂了），或執(zhí)行了足夠多的迭代
算法的時(shí)間復(fù)雜度上界為O(n*k*t), 其中k為輸入的聚類個(gè)數(shù)，n為數(shù)據(jù)量，t為迭代次數(shù)。一般t,k,n均可認(rèn)為是常量，時(shí)間和空間復(fù)雜度可以簡(jiǎn)化為O(n)，即線性的

spark ml編碼實(shí)踐

可在spark-shell環(huán)境下修改參數(shù)調(diào)試以下代碼，可以用實(shí)際的業(yè)務(wù)數(shù)據(jù)做測(cè)試評(píng)估，業(yè)務(wù)數(shù)據(jù)一般是多列，可以把維度列用VectorAssembler組裝成向量列做為Kmeans算法的輸入列，考慮現(xiàn)實(shí)的應(yīng)用場(chǎng)景，比如做異常數(shù)據(jù)檢測(cè)，正常數(shù)據(jù)分為一類，異常數(shù)據(jù)分為幾類，分別統(tǒng)計(jì)正常數(shù)據(jù)與異常數(shù)據(jù)的數(shù)據(jù)量，求百分比等

<span style="font-size:18px;">import org.apache.spark.ml.clustering.KMeans  
    import org.apache.spark.mllib.linalg.Vectors  
      
    val dataset = sqlContext.createDataFrame(Seq(  
    (1, Vectors.dense(0.0, 0.0, 0.0)),  
    (2, Vectors.dense(0.1, 0.1, 0.1)),  
    (3, Vectors.dense(0.2, 0.2, 0.2)),  
    (4, Vectors.dense(9.0, 9.0, 9.0)),  
    (5, Vectors.dense(1.1, 1.1, 0.1)),  
    (6, Vectors.dense(12, 14, 100)),  
    (6, Vectors.dense(1.1, 0.1, 0.2)),  
    (6, Vectors.dense(-2, -3, -4)),  
    (6, Vectors.dense(1.6, 0.6, 0.2))  
    )).toDF("id", "features")  
      
    // Trains a k-means model  
    val kmeans = new KMeans().setK(3).setMaxIter(20).setFeaturesCol("features").setPredictionCol("prediction")  
    val model = kmeans.fit(dataset)  
      
    // Shows the result  
    println("Final Centers: ")  
    model.clusterCenters.foreach(println)  
    model.clusterCenters.zipWithIndex.foreach(println)  
      
    val myres = model.transform(dataset).select("features","prediction")  
    myres.show()</span>
聚類算法是一類無監(jiān)督式機(jī)器學(xué)習(xí)算法，聚類效果怎么評(píng)估，模型訓(xùn)練參數(shù)怎么調(diào)優(yōu)，是否能用管道來訓(xùn)練模型來比較各種不同組合的參數(shù)的效果，即網(wǎng)格搜索法(grid search),先設(shè)置好待測(cè)試的參數(shù)，MLLib就會(huì)自動(dòng)完成這些參數(shù)的不同組合,管道搭建了一條工作流，一次性完成了整個(gè)模型的調(diào)優(yōu)，而不是獨(dú)立對(duì)每個(gè)參數(shù)進(jìn)行調(diào)優(yōu)，這個(gè)還要再確認(rèn)一下，查看SPARK-14516好像目前還沒有一個(gè)聚類效果通用的自動(dòng)的度量方法
像這種代碼（不過現(xiàn)在這個(gè)代碼有問題）：

<span style="font-size:18px;">import org.apache.spark.ml.clustering.KMeans  
    import org.apache.spark.mllib.linalg.Vectors  
    import org.apache.spark.ml.tuning.{ ParamGridBuilder, CrossValidator }  
    import org.apache.spark.ml.{ Pipeline, PipelineStage }  
      
    val dataset = sqlContext.createDataFrame(Seq(  
    (1, Vectors.dense(0.0, 0.0, 0.0)),  
    (2, Vectors.dense(0.1, 0.1, 0.1)),  
    (3, Vectors.dense(0.2, 0.2, 0.2)),  
    (4, Vectors.dense(9.0, 9.0, 9.0)),  
    (5, Vectors.dense(1.1, 1.1, 0.1)),  
    (6, Vectors.dense(12, 14, 100)),  
    (6, Vectors.dense(1.1, 0.1, 0.2)),  
    (6, Vectors.dense(-2, -3, -4)),  
    (6, Vectors.dense(1.6, 0.6, 0.2))  
    )).toDF("id", "features")  
      
    val kmeans = new KMeans().setK(2).setMaxIter(20).setFeaturesCol("features").setPredictionCol("prediction")  
    //主要問題在這里，沒有可用的評(píng)估器與label列設(shè)置  
    val evaluator = new BinaryClassificationEvaluator().setLabelCol("prediction")  
    val paramGrid = new ParamGridBuilder().addGrid(kmeans.initMode, Array("random")).addGrid(kmeans.k, Array(3, 4)).addGrid(kmeans.maxIter, Array(20, 60)).addGrid(kmeans.seed, Array(1L, 2L)).build()  
    val steps: Array[PipelineStage] = Array(kmeans)  
    val pipeline = new Pipeline().setStages(steps)  
      
    val cv = new CrossValidator().setEstimator(pipeline).setEvaluator(evaluator).setEstimatorParamMaps(paramGrid).setNumFolds(10)  
    // Trains a model  
    val pipelineFittedModel = cv.fit(dataset)</span>