1. 背景

1.1 问题概述

有10万+条短文本，均是用户反馈的问题（每条文本长度大概在200字左右），需要对这些文本进行主题聚类，看下用户反馈的问题都集中在哪些方面。

1.2 工作

先采用Spark MLlib自带的Kmeans聚类算法对文本进行聚类，因为其训练速度很快。这里我采用TF-IDF作为特征提取方法，Spark ML Kmeans中的距离计算方法为欧式距离。
运行之后发现效果不太好：某个类的数据量达到了70%，也就是大量的文章都被划分到了同一组，其他类的数量都较少。

如下图所示，对欧式距离的Kmeans和其他距离方法进行了对比，实验中表明欧式距离的结果中，本应属于不同组的大量文章被划分成了一组，与我这边的效果一致，因此接下来可以试着采用余弦相似度作为距离算法。

欧式距离和余弦相似度的实验效果，转载(http://xueshu.baidu.com/usercenter/paper/show?paperid=a3195f1409270d32f304145ce00e967e&site=xueshu_se)

2. 第一步优化：欧式距离改为余弦相似度

2.1 欧式距离和余弦相似度对比

定义两个n维向量：X(x1,x2,...,xn)和Y(y1,y2,...,yn)
欧式距离计算公式：

欧式距离

余弦相似度计算公式：

余弦相似度

欧式距离主要是衡量空间中两个点的绝对距离，而余弦相似度注重两个样本之间在方向上的差异而非距离上的差异，主要是衡量两个个体之间的相似性，值越大，说明差异越少，与欧式距离相反（距离越小，差异越小）。

欧式距离和余弦相似度对比

从上图可以看出，欧氏距离衡量的是空间各点的绝对距离；而余弦距离衡量的是空间向量的夹角，更加体现在方向上的差异，而不是位置。如果保持 A 点位置不变，B 点朝原方向远离坐标轴原点，那么这个时候余弦距离是保持不变的（因为夹角没有发生变化），而 A 与 B 两点的距离显然在发生变化，这就是欧式距离与余弦相似度的不同之处。

2.2 开发基于Spark和余弦相似度的Kmeans聚类

由于Spark ML中的Kmeans不提供对距离函数进行更新和选择的接口，因此只能按照Kmeans的原理开发，和在GitHub上借鉴已有的代码修改开发。
这里有个小技巧，由于余弦相似度越大，两个体之间的差异越少，所以为了保证在计算每个样本所属的最近的中心点的时候与欧式距离一致，这里在计算两个体之间距离的时候采用如下方法计算：
（代码框架参考基于欧式距离的Scala实现的Kmeans，基于该代码修改为Spark和余弦相似度距离。https://blog.csdn.net/u014135021/article/details/53668634）

 /**求两个向量的余弦,1-相似度，结果越大 差异越大，越小差异越小 */
  def cos_distance(that: Point) = {
    val cos = 1- innerProduct(this.px, that.px) / (module(this.px) * module(that.px))
    cos
  }
 /** 求两个向量的内积*/
  def innerProduct(v1: Vector[Double], v2: Vector[Double]) = {
    val listBuffer = ListBuffer[Double]()    
    for (i <- 0 until v1.length; j <- 0 to v2.length; if i == j) {
      if (i == j) listBuffer.append(v1(i) * v2(j))
    }
    listBuffer.sum
  }
  

3. 第二步优化：Scala代码改成Spark

如下为Spark的Kmeans主题的聚类迭代部分
其中初始化随机中心点的方法为：

takeSample(withReplacement: Boolean,num: Int,seed: Long = Utils.random.nextLong): Array[T]

其中参数：

1. withReplacement：是否是有放回的抽样
2. num：返回的样本的大小
3. seed：随机数生成器的种子

  //kmeans函数运行主体
  def run(sc:SparkContext)
  {
  
    var k=0 //当前迭代次数
    var f=true //是否还需要接着迭代
    val st=System.nanoTime()
    //设置随机种子
    val seed = 10000l
    val random = new java.util.Random()
    random.setSeed(seed)
    InitCenterRandom(random)//随机初始化中心点
    while(k<MaxIterations && f)
    {
      val st1 = System.currentTimeMillis()
      k+=1
      //计算每个点属于哪个中心点所在的类，并且记录每个类中点的数量，与该类中所有向量的和        
      val data_with_center = data.map(x => {
        var cid = FastSearch2(x._2).center_id
        (x._1,x._2,cid)
      })
      data_with_center.cache()
      //按照类别ID分组
      val result_groupby:RDD[(Int, Iterable[(X,Point,Int)])] = data_with_center.groupBy(_._3) 
      result_groupby.cache()
      result = result_groupby.map(x => {
        val center_datas = x._2.map(_._1).toList
        (x._1,center_datas)
      })
      //按照中心点相同groupby
      val newPoints = result_groupby.map(x => {
        val cid = x._1
        val center_datas = x._2
        val center_data_size:Int = center_datas.seq.size.toInt
        //计算该中心点下所有数据的Point向量和
        val totalPoint:Point = center_datas.map(_._2).reduce((x,y) => (x+y))
        //新的中心点为该类别下样本向量和的平均值
        val newPoint:Point = totalPoint./(center_data_size)
        (cid, newPoint)
      })
      newPoints.cache()
      val newPoints2 = newPoints.collect()
      result_groupby.unpersist()
      data_with_center.unpersist()
      var i = -1
      //如果当前中心点中，存在比上一次迭代的中心点的距离大于阈值的情况，还需要接着迭代。
      f = CenterPoint.map(x => {
        i = i+1
        (i, x)
      }).zip(newPoints2).map(f=>f._1._2.cos_distance(f._2._2)).exists {_>threshold}
      
      //更新中心点
      if(f)
      {
        newPoints2.map(x=> {
          if(x._2 != null) {
            CenterPoint(x._1) = x._2
          }
        })
      }
      val et1 = System.currentTimeMillis()
      println("第"+k+"次聚类,sse=" + getSSE(sc,data) + ",time=" + (et1-st1)/1000+"s")
      newPoints.unpersist()
      System.gc()
      
    }
    val ed=System.nanoTime()
    //data.unpersist()
    println("Kmeans聚类时间为:"+(ed-st))

  }

/*根据随机种子对象，初始化中心*/
def InitCenterRandom(random:java.util.Random) {
    val st=System.nanoTime()    
    val random_seed = random.nextLong()
    CenterPoint = data.takeSample(false, numClusters, random_seed).map(_._2)
    val ed=System.nanoTime()
    println("随机中心点生成时间为："+(ed-st))
  }

4. 第三步优化：大数据量的Mini Batch Kmeans

上述的Kmeans算法在大数据量的情况下，运算依然很慢，因此采用KMeans的变种：Mini Batch Kmeans算法，当数据量超过1万的时候就可以使用该方法。该方法不仅处理速度快，准确度也很高。其实现原理是每次迭代的时候，选取部分样本来更新当前迭代的中心点。这种分批处理的思路同样也被应用在了梯度下降等算法中。

如下图为摘自链接：https://blog.csdn.net/cht5600/article/details/76014573
将Kmeans算法与Mini Batch Kmeans算法的聚类结果对比，第三幅图代表两种方式分类差异的样本：

Kmeans & MiniBatchKMeans

从图中可以知道，针对同样数量的文本分别采用Kmeans和Mini Batch Kmeans训练，其时间差别较大，且inertia相差较少。

inertia：样本离最近聚类中心的总和,其是K均值模型对象的属性，表示样本距离最近的聚类中心的总和，它是作为在没有真实分类标签下的非监督式评估指标，该值越小越好，值越小证明样本在类间的分布越集中，即类内的距离越小。

因此，该方法在尽量保持准确度的情况下，大大减少了聚类时间。在本次实验中采用10万样本，选取1000维特征，每次选取1000个样本迭代，每次迭代仅需要1分钟。
如下为Mini Batch Kmeans的迭代部分,大部分逻辑与上面的Kmeans一致，只有在每次迭代的选取的样本不同：

//kmeans函数运行主体
  def runBatch(sc:SparkContext)
  {
    
    var k=0
    var f=true
    val st=System.nanoTime()
    //设置随机种子
    val seed = 10000l
    val random = new java.util.Random()
    random.setSeed(seed)
    InitCenterRandom(random)//随机初始化中心点

    while(k<MaxIterations && f)
    {
      val st1 = System.currentTimeMillis()
      k+=1
      //堆积选取MiniBatchSize个样本 转成RDD
      val data_batch = sc.parallelize(data.takeSample(false,ConfigUtil.MiniBatchSize,random.nextLong()))
      data_batch.cache()
      val data_with_center = data_batch.map(x => {
        var cid = FastSearch2(x._2).center_id
        (x._1,x._2,cid)
      })
      data_with_center.cache()
     val result_groupby:RDD[(Int, Iterable[(GovComment,Point,Int)])] = data_with_center.groupBy(_._3)
      result_groupby.cache()
      result = result_groupby.map(x => {
        val center_datas = x._2.map(_._1).toList
        (x._1,center_datas)
      })
      //按照中心点相同groupby
      val newPoints = result_groupby.map(x => {
        val cid = x._1
        val center_datas = x._2
        val center_data_size:Int = center_datas.seq.size.toInt
        //计算该中心点下所有数据的Point向量和
        val totalPoint:Point = center_datas.map(_._2).reduce((x,y) => (x+y))
        val newPoint:Point = totalPoint./(center_data_size)
        (cid, newPoint)
      })
      newPoints.cache()
      val newPoints2 = newPoints.collect()
      result_groupby.unpersist()
      data_with_center.unpersist()
      var i = -1
      f = CenterPoint.map(x => {
        i = i+1
        (i, x)
      }).zip(newPoints2).map(f=>f._1._2.cos_distance(f._2._2)).exists {_>threshold}

      if(f)
      {
        newPoints2.map(x=> {
          if(x._2 != null) {
            CenterPoint(x._1) = x._2
          }
        })
      }
      println("第"+k+"次聚类,sse=" + getSSE(sc,data_batch) + ",time=" +(System.currentTimeMillis()-st1)/1000+"s")
      newPoints.unpersist()
      System.gc()
      data_batch.unpersist()
    }
    val ed=System.nanoTime()
    //data.unpersist()
    println("Kmeans聚类时间为:"+(ed-st))

  }

5. 总结

1. 针对短文本聚类，可以在条件允许的情况下提高特征维度；
2. Spark Accumulator累加器的使用注意；
3. 采用Mini Batch Kmeans可以尽量维持聚类准确度；
4. 文本上面相似，余弦相似度效果相对欧式距离好些。

6. 参考

1. https://blog.csdn.net/huangfei711/article/details/78469614
2. http://xueshu.baidu.com/usercenter/paper/show?paperid=a3195f1409270d32f304145ce00e967e&site=xueshu_se
3. https://blog.csdn.net/linvo/article/details/9333019
4. https://blog.csdn.net/u014135021/article/details/53668634
5. https://blog.csdn.net/cht5600/article/details/76014573
6. https://blog.csdn.net/weixin_37536446/article/details/81326932