spark機器學習-聚類

聚類算法是一種無監督學習任務，用于將對象分到具有高度相似性的聚類中，聚類算法的思想簡單的說就是物以類聚的思想，相同性質的點在空間中表現的較為緊密和接近，主要用于數據探索與異常檢測，最常用的一種聚類算法是K均值(K-means)聚類算法

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

spark ml編碼實踐

可在spark-shell環境下修改參數調試以下代碼，可以用實際的業務數據做測試評估，業務數據一般是多列，可以把維度列用VectorAssembler組裝成向量列做為Kmeans算法的輸入列，考慮現實的應用場景，比如做異常數據檢測，正常數據分為一類，異常數據分為幾類，分別統計正常數據與異常數據的數據量，求百分比等

<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>
聚類算法是一類無監督式機器學習算法，聚類效果怎么評估，模型訓練參數怎么調優，是否能用管道來訓練模型來比較各種不同組合的參數的效果，即網格搜索法(grid search),先設置好待測試的參數，MLLib就會自動完成這些參數的不同組合,管道搭建了一條工作流，一次性完成了整個模型的調優，而不是獨立對每個參數進行調優，這個還要再確認一下，查看SPARK-14516好像目前還沒有一個聚類效果通用的自動的度量方法
像這種代碼（不過現在這個代碼有問題）：

<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")  
    //主要問題在這里，沒有可用的評估器與label列設置  
    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>