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