Spark2.0機器學習系列之11： 聚類(冪迭代聚類， power iteration clustering， PIC)

在Spark2.0版本中（不是基於RDD API的MLlib），共有四種聚類方法： 

            （1）K-means 
            （2）Latent Dirichlet allocation (LDA) 
            （3）Bisecting k-means（二分k均值算法） 
            （4）Gaussian Mixture Model (GMM)。 
            基於RDD API的MLLib中，共有六種聚類方法： 
            （1）K-means 
            （2）Gaussian mixture 
            （3）Power iteration clustering (PIC) 
            （4）Latent Dirichlet allocation (LDA)** 
            （5）Bisecting k-means 
            （6）Streaming k-means 
            多了Power iteration clustering (PIC)和Streaming k-means兩種。 
           本文將PIC,即冪迭代聚類。其它方法在我Spark機器學習系列裏面都有介紹。

概述

           譜聚類（Spectral Clustering）相信你們可能很是熟悉（若是不熟悉的話相關資料也比較多），而冪迭代聚類（Power iteration clustering）則也許會比較陌生。其實這二者之間有不少類似的地方，可是算法仍是有較大差別的，這些後面會慢慢道來。 
            冪迭代聚類來自Frank Lin 和William W. Cohen這兩位卡內基梅隆大學大牛，發表於ICML 2010。深刻了解算法則須要看看大牛的原文。 
            這裏給出連接http://www.cs.cmu.edu/~wcohen/postscript/icml2010-pic-final.pdf 
           若是具有必定的譜聚類和Graphx圖計算的知識，就比較容易理解冪迭代聚類，不過不太懂也沒有關係，Graphx圖計算能夠簡單看看個人Spark系列博文，譜聚類後面會簡單介紹一下，另外算法的數學基礎是冪迭代求特徵值，我後面會詳細介紹。 
           兩位大牛提出一種簡單且可擴展的圖聚類方法，稱之爲冪迭代聚類（PIC）。在數據歸一化的逐對類似矩陣上，使用截斷的冪迭代，PIC尋找數據集的一個超低維嵌入（低緯空間投影，embedding ），這種嵌入剛好是頗有效的聚類指標，使它在真實數據集上老是好於普遍使用的譜聚類方法（好比說NCut）。PIC在大數據集上很是快，比基於當前（2010年）最好的特徵向量計算技術實現的NCut還要快1000倍。 
       We present a simple and scalable graph clustering method called power iteration clustering (PIC). PIC finds a very low-dimensional embedding of a dataset using truncated power iteration on a normalized pair-wise similarity matrix of the data. This embedding turns out to be an effective cluster indicator, consistently outperforming widely used spectral methods such as NCut on real datasets. PIC is very fast on large datasets, running over 1,000 times faster than an NCut implementation based on the state-of-the-art IRAM eigenvector computation technique.

冪迭代法求矩陣的主特徵值

           首先仍是理清基本的數學算法，這樣後面分析就容易多了。「冪迭代」法求特徵值，也有直接就叫作「冪法」求特徵值的，也是最基礎的一種特徵值迭代法求解方法。 
           適合計算大型稀疏矩陣的主特徵值，即按模最大的特徵值，同時也獲得了對應的特徵向量（這不就是爲大數據集，一般仍是稀疏矩陣量身打造的嗎？呵呵）。 
        它的優勢是方法簡單，理論依據是迭代的收斂性（這兩點要看完下面的過程才能深入的理解）

        冪法基本思想是:若咱們求某個n階方陣A的特徵值和特徵向量,先任取一個非零初始向量v(0),進行以下的迭代計算,直到收斂（下面都是對｛v(k)，   k=0,1,...｝序列而言的）： 

 

 

 若是直接出給份量表達式，那麼誰看得懂這個公式怎麼來的，j什麼意思？因此簡單看看文獻【6】，內心就踏實多了，是分多種狀況的，你們直接連接吧，有點複雜，我就不細說了。 
       因此咱們能看出，方法確實簡單，很是容易就能夠寫一個能運行的c/c++程序，方法的理論依據就是迭代的收斂性。利用了模大小不一樣的各特徵值在迭代中隨着迭代次數的增長貢獻差別愈發明顯這個特性，舉個不恰當例子，就比如選一羣馬來長跑，其中有一匹特別能跑，並且耐力也特別好，是一匹千里馬，一開有必定差距，慢慢的就把其它馬落的愈來愈來，以致於最後你們忽略了其它的馬，只看這以匹就行。 
       以上就是冪迭代求λ1的基本原理，至於對收斂速度的分析，以及各種加速方法，就再也不繼續了，敲公式太累了。。。（其實省略的這些內容從公式推理上來講是比較簡單的！！！）

 

譜聚類

能夠直接連接：http://www.tuicool.com/articles/Nzumuu 
下面的資源也對學習頗有幫助：http://www.doc88.com/p-5408063029057.html 
       譜聚類(Spectral Clustering, SC)是一種基於圖論的聚類方法——將帶權無向圖劃分爲兩個或兩個以上的最優子圖，使子圖內部儘可能類似，而子圖間距離儘可能距離較遠，以達到常見的聚類的目的。其中的最優是指最優目標函數不一樣，能夠是割邊最小分割——如圖1的Smallest cut(如後文的Min cut)， 也能夠是分割規模差很少且割邊最小的分割——Best cut(如後文的Normalized cut)。 
        譜聚類可以識別任意形狀的樣本空間且收斂於全局最優解，其基本思想是利用樣本數據的類似矩陣L(拉普拉斯矩陣)進行特徵分解後獲得的特徵向量進行聚類。 
       給定樣本的原始特徵，咱們須要計算兩兩樣本之間的類似度值，才能構造出關聯矩陣A。咱們通常使用高斯核函數來計算類似度。 
        拉普拉斯矩陣  L=D−A  , 其中D爲圖的度矩陣，度是圖論中的概念，也就是矩陣A行或列的元素之和。譜聚類基於矩陣L來進行研究。

冪迭代聚類

       假設有數據集X=x1,x2,...,xn,類似函數s(xi,xj)。定義關聯矩陣A,並抽象成圖。（這部分和譜聚類是同樣的）。 
       歸一化的關聯矩陣W定義爲： 

Spark MLlib 代碼

           應該說Spark中Graphx 模塊的發展，極大促進了基於圖計算的諸多算法在大數據集上的實現（關於Graphx的內容，也能夠參考我Spark系列的博文），譜聚類方法自己就是一種基於圖的技術，冪迭代聚類和譜聚類同樣，也是基於圖的聚類技術，都是先把得到的原始數據集抽象爲圖結構後（頂點，邊和三元組等等）才進一步進行分析的。

           程序關鍵參數是確保輸入的圖RDD符合無向，非負等多個要求，這在下面程序註釋中都已經一一說明了。 
           另外K值的選取和k-means方法同樣，仍然是一個難題，能夠參看一下我Spark系列文章k-means那篇。

import org.apache.spark.sql.SparkSession
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.mllib.clustering.PowerIterationClustering


object myClusters2 {

  //產生一個分佈在圓形上的數據模型（見程序後面的圖）：參數爲半徑和點數
  def generateCircle(radius: Double, n: Int): Seq[(Double, Double)] = {
    Seq.tabulate(n) { i =>
      val theta = 2.0 * math.Pi * i / n
      (radius * math.cos(theta), radius * math.sin(theta))
    }
  } 

  //產生同心圓樣的數據模型（見程序後面的圖）：參數爲同心圓的個數和點數
  //第一個同心圓上有nPoints個點，第二個有2*nPoints個點，第三個有3*nPoints個點，以此類推
  def generateCirclesRdd(
      sc: SparkContext,
      nCircles: Int,
      nPoints: Int): RDD[(Long, Long, Double)] = {
    val points = (1 to nCircles).flatMap { i =>
      generateCircle(i, i * nPoints)
    }.zipWithIndex
    val rdd = sc.parallelize(points)
    val distancesRdd = rdd.cartesian(rdd).flatMap { case (((x0, y0), i0), ((x1, y1), i1)) =>
      if (i0 < i1) {
        Some((i0.toLong, i1.toLong, gaussianSimilarity((x0, y0), (x1, y1))))
      } else {
        None
      }
    }
    distancesRdd
  }

  /**
   * Gaussian Similarity
   */
  def gaussianSimilarity(p1: (Double, Double), p2: (Double, Double)): Double = {
    val ssquares = (p1._1 - p2._1) * (p1._1 - p2._1) + (p1._2 - p2._2) * (p1._2 - p2._2)
    math.exp(-ssquares / 2.0)
  }


  def main(args:Array[String]){

    //屏蔽日誌
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    val warehouseLocation = "/Spark/spark-warehouse"

    val spark=SparkSession
            .builder()
            .appName("myClusters")
            .master("local[4]")            
            .config("spark.sql.warehouse.dir",warehouseLocation)            
            .getOrCreate();       

    val sc=spark.sparkContext

    // 產生測試數據
    val circlesRdd = generateCirclesRdd(sc, 2, 4)//產生具備2個半徑不一樣的同心圓模型的類似度矩陣，第一個圓上有4個點，第二個圓上有2*4=8個點    
    //circlesRdd.take(100).foreach(p=>print(p+"\n"))
    /**
     * circlesRdd數據格式實例：(srcId, dstId, similarity)
     * (0,1,0.22313016014842987)    
     * (0,2,0.22313016014842973)     
     * (1,2,0.22313016014842987)    
     */

    /**PowerIterationClustering函數輸入輸出說明
     * 
     *  It takes an RDD of (srcId, dstId, similarity) tuples and outputs a model
     *  with the clustering assignments. The similarities must be nonnegative. 
     *  PIC assumes that the similarity measure is symmetric. A pair (srcId, dstId) 
     *  regardless of the ordering should appear at most once in the input data. 
     *  If a pair is missing from input, their similarity is treated as zero. 
     *  它的輸入circlesRdd是(srcId, dstId,similarity) 三元組的RDD（起始點->終點，類似度爲邊權重）。
     *  顯然算法中選取的類似度函數是非負的，並且PIC 算法假設類似函數還知足對稱性（邊權重表明類似度，非負）。
     *  輸入數據中（srcId,dstId）對無論前後順序如何只能出現至多一次（便是無向圖，兩頂點邊權重是惟一的）
     *  若是輸入數據中不含這個數據對（不考慮順序），則這兩個數據對的類似度爲0（沒有這條邊）.
     *  
     */
    val modelPIC = new PowerIterationClustering()
      .setK(2)// k : 指望聚類數 
      .setMaxIterations(40)//冪迭代最大次數
      .setInitializationMode("degree")//模型初始化，默認使用」random」 ，即便用隨機向量做爲初始聚類的邊界點，能夠設置」degree」（就是圖論中的度）。
      .run(circlesRdd)


    //輸出聚類結果
    val clusters = modelPIC.assignments.collect().groupBy(_.cluster).mapValues(_.map(_.id))
    val assignments = clusters.toList.sortBy { case (k, v) => v.length }
    val assignmentsStr = assignments
      .map { case (k, v) =>
        s"$k -> ${v.sorted.mkString("[", ",", "]")}"
      }.mkString(", ")
    val sizesStr = assignments.map {
      _._2.length
    }.sorted.mkString("(", ",", ")")
    println(s"Cluster assignments: $assignmentsStr\ncluster sizes: $sizesStr")  
    /*
     * Cluster assignments: 1 -> [4,6,8,10], 0 -> [0,1,2,3,5,7,9,11]
     * cluster sizes: (4,8)
     */   

  }  

}

 
代碼中產生的模型

參考文獻 
（1）Spark官網 
（2)《power iteration clustering》 Lin and Cohen 
http://citeseerx.ist.psu.edu/viewdoc/download?doi=10.1.1.148.2474&rep=rep1&type=pdf 
（3）Spectral Clustering via the Power Method – Provably 
Christos Boutsidis,Alex Gittens,Prabhanjan Kambadur 
（4）百度文庫 xiaofei id 
http://wenku.baidu.com/link?url=uSwQhuvB9emiBBNk7tYqHE8KP6Knt6ydvaYBqIlnFL-CVdIBaTTpiV1kOWJPOTMHL9WsbExg07sT566eI2NEUUGV32mFsaGGjx13TuqTu4q 
（5）維基 https://en.wikipedia.org/wiki/Power_iteration 
（6）冪法求矩陣特徵值的一些補充 何 明 安徽大學計算機科學與工程系 合肥 230039 
http://wenku.baidu.com/link?url=HdDgi9ypgjhGrAa65mQCTeX1XlgFRZ7qUIssSQ-3zSFp6voO3Bd9YnP5j1_mri5clfLPCgCNQk-LMvPszzezkF-alfhp4iGe42gQhOwTxPa