首页 > 代码库 > 五、聚类

五、聚类

聚类:
聚类是一个无监督学习问题,我们基于相似的特性将数据分组成多个子集。聚类通常用于探索性分析或者作为分层监督学习管道(每个簇训练不同的分类或者回归模型)的组件。
 
MLlib支持下面的几个模型:
 
K均值(K-means)
高斯混合(Gaussian mixture)
幂迭代聚类(Power iteration clustering (PIC))
隐含狄利克雷分布(Latent Dirichlet allocation (LDA))
流式K均值(Streaming k-means)
 
K均值(k-means)是最通用的聚类算法之一,该算法将数据点聚类为指定数量的簇(注:基本算法原理是随机挑选N个中心点,每轮计 算所有点到中心点的距离,并将点放到最近的中心,然后均值更新中心点,然后重复上述过程直至收敛,收敛的判断依据是距离阈 值)。MLLib的实现包含了 k-means++的并行计算变体,该算法也叫kmeans||。它有下列参数:
 
k 需要聚簇的数量
maxIterations 最大迭代次数
initializationMode 指定初始化的模式,可以是随机初始化也可以是k-means||初始化 (k-means||初始化不全是随机选点,而是使用一个算法使选的点尽可能分散).
runs 执行K均值聚簇算法的次数  (k-means不保证能找到全局最优解,同一数据集上执行多次的话,可以返回更好的聚簇结果)。
initializationSteps 使用k-means|| 算法选初始点时最多迭代的次数.
epsilon 判定k-means是否收敛的距离阈值(聚簇中心前后两次的差值小于epsilon即达到收敛条件)
initialModel 初始模型的聚类中心用于初始化一个可选设置。如果提供此参数,只进行一次
 
补充1:kmeans的损失函数。其中(x1, x2, …, xn)是点集,每个点是d维向量,S是聚类的k个簇,μi 是Si 中所有点的均值)。这 个损失函数也叫WSSS( within set sum of square)
 
技术分享
 
补充2:kmeans++方法:
 
kmeans++算法的主要工作体现在种子点的选择上,基本原则是使得各个种子点之间的距离尽可能的大,但是又得排除噪声的影响。
以下为基本思路:[1]
1、从输入的数据点集合(要求有k个聚类)中随机选择一个点作为第一个聚类中心
2、对于数据集中的每一个点x,计算它与最近聚类中心(指已选择的聚类中心)的距离D(x)
3、选择一个新的数据点作为新的聚类中心,选择的原则是:D(x)较大的点,被选取作为聚类中心的概率较大
4、重复2和3直到k个聚类中心被选出来
5、利用这k个初始的聚类中心来运行标准的k-means算法
 
import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.clustering.KMeans;
import org.apache.spark.mllib.clustering.KMeansModel;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.SparkConf;

public class KMeansExample {
  public static void main(String[] args) {
    SparkConf conf = new SparkConf().setAppName("K-means Example");
    JavaSparkContext sc = new JavaSparkContext(conf);

    // Load and parse data
    String path = "data/mllib/kmeans_data.txt";
    JavaRDD<String> data =http://www.mamicode.com/ sc.textFile(path);
    JavaRDD<Vector> parsedData =http://www.mamicode.com/ data.map(
      new Function<String, Vector>() {
        public Vector call(String s) {
          String[] sarray = s.split(" ");
          double[] values = new double[sarray.length];
          for (int i = 0; i < sarray.length; i++)
            values[i] = Double.parseDouble(sarray[i]);
          return Vectors.dense(values);
        }
      }
    );
    parsedData.cache();

    // Cluster the data into two classes using KMeans
    int numClusters = 2;
    int numIterations = 20;
    KMeansModel clusters = KMeans.train(parsedData.rdd(), numClusters, numIterations);

    // Evaluate clustering by computing Within Set Sum of Squared Errors
    double WSSSE = clusters.computeCost(parsedData.rdd());
    System.out.println("Within Set Sum of Squared Errors = " + WSSSE);

    // Save and load model
    clusters.save(sc.sc(), "myModelPath");
    KMeansModel sameModel = KMeansModel.load(sc.sc(), "myModelPath");
  }
}

 

高斯混合 [Gaussian mixture]
高斯混合模型 表达的是一种混合分布,所有点都来自于k个高斯子分布中的一个,每个点都对应一个相应的概率。在MLlib的实现中 ,对于给定的样本集,使用最大期望算法(EM)来引导最大似然模型。算法实现由下列参数:
 
k 目标聚簇数量
convergenceTol 两次迭代损失(log-likelihood)变化的容忍度.
maxIterations 收敛之前可以运行的最大迭代次数
初始模型是一个可选的出发点从开始的EM算法。如果省略这个参数,将从数据中构造一个随机起点.。
 
import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.clustering.GaussianMixture;
import org.apache.spark.mllib.clustering.GaussianMixtureModel;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.SparkConf;

public class GaussianMixtureExample {
  public static void main(String[] args) {
    SparkConf conf = new SparkConf().setAppName("GaussianMixture Example");
    JavaSparkContext sc = new JavaSparkContext(conf);

    // Load and parse data
    String path = "data/mllib/gmm_data.txt";
    JavaRDD<String> data =http://www.mamicode.com/ sc.textFile(path);
    JavaRDD<Vector> parsedData =http://www.mamicode.com/ data.map(
      new Function<String, Vector>() {
        public Vector call(String s) {
          String[] sarray = s.trim().split(" ");
          double[] values = new double[sarray.length];
          for (int i = 0; i < sarray.length; i++)
            values[i] = Double.parseDouble(sarray[i]);
          return Vectors.dense(values);
        }
      }
    );
    parsedData.cache();

    // Cluster the data into two classes using GaussianMixture
    GaussianMixtureModel gmm = new GaussianMixture().setK(2).run(parsedData.rdd());

    // Save and load GaussianMixtureModel
    gmm.save(sc.sc(), "myGMMModel");
    GaussianMixtureModel sameModel = GaussianMixtureModel.load(sc.sc(), "myGMMModel");
    // Output the parameters of the mixture model
    for(int j=0; j<gmm.k(); j++) {
        System.out.printf("weight=%f\nmu=%s\nsigma=\n%s\n",
            gmm.weights()[j], gmm.gaussians()[j].mu(), gmm.gaussians()[j].sigma());
    }
  }
}

 

幂迭代聚类 (PIC)
对于图的顶点聚类(顶点相似度作为边的属性)问题,幂迭代聚类(PIC)是高效并且易扩展的算法(参考: Lin and Cohen, Power Iteration Clustering)。MLlib包含了一个使用GraphX(MLlib)为基础的实现。算法的输入是RDD[srcID, dstID, similarity],输出是每个顶点对应的聚类的模型。相似度(similarity)必须是非负值。PIC假设相似度的衡量是对称的,也就是说在输入数据中,(srcID, dstID)顺序无关(例如:<1, 2, 0.1>, <2, 1, 0.1等价),但是只能出现一次。输入中没有指定相似度的点对,相似度会置0。MLlib中的PIC实现具有下列参数:
 
k:  聚簇的数量
maxIterations: 最大迭代次数
initializationMode: 初始化模式:默认值“random”,表示使用一个随机向量作为顶点的聚类属性;也可以是“degree”,表示使用归一化的相似度和(作为顶点的聚类属性)。
 
import scala.Tuple2;
import scala.Tuple3;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.clustering.PowerIterationClustering;
import org.apache.spark.mllib.clustering.PowerIterationClusteringModel;

// Load and parse the data
JavaRDD<String> data = http://www.mamicode.com/sc.textFile("data/mllib/pic_data.txt");
JavaRDD<Tuple3<Long, Long, Double>> similarities = data.map(
  new Function<String, Tuple3<Long, Long, Double>>() {
    public Tuple3<Long, Long, Double> call(String line) {
      String[] parts = line.split(" ");
      return new Tuple3<>(new Long(parts[0]), new Long(parts[1]), new Double(parts[2]));
    }
  }
);

// Cluster the data into two classes using PowerIterationClustering
PowerIterationClustering pic = new PowerIterationClustering()
  .setK(2)
  .setMaxIterations(10);
PowerIterationClusteringModel model = pic.run(similarities);

for (PowerIterationClustering.Assignment a: model.assignments().toJavaRDD().collect()) {
  System.out.println(a.id() + " -> " + a.cluster());
}

// Save and load model
model.save(sc.sc(), "myModelPath");
PowerIterationClusteringModel sameModel = PowerIterationClusteringModel.load(sc.sc(), "myModelPath");

 

隐含狄利克雷分布 (LDA)
隐含狄利克雷分布(LDA) 是一个主题模型,它能够推理出一个文本文档集合的主体。LDA可以认为是一个聚类算法,原因如下:
 
主题对应聚类中心,文档对应数据集中的样本(数据行)
主题和文档都在一个特征空间中,其特征向量是词频向量。
跟使用传统的距离来评估聚类不一样的是,LDA使用评估方式是一个函数,该函数基于文档如何生成的统计模型。
 
LDA以词频向量表示的文档集合作为输入。然后在最大似然函数上使用期望最大(EM)算法 来学习聚类。完成文档拟合之后,LDA提
供:
 
Topics: 推断出的主题,每个主体是单词上的概率分布。
Topic distributions for documents: 对训练集中的每个文档,LDA给了一个在主题上的概率分布。
 
LDA参数如下:
 
k:  主题数量(或者说聚簇中心数量)
maxIterations: EM算法的最大迭代次数。
docConcentration: 文档在主题上分布的先验参数。当前必须大于1,值越大,推断出的分布越平滑。
topicConcentration: 主题在单词上的先验分布参数。当前必须大于1,值越大,推断出的分布越平滑。
checkpointInterval: 检查点间隔。maxIterations很大的时候,检查点可以帮助减少shuffle文件大小并且可以帮助故障恢复。
 
注意:当前在MLlib中,LDA是一个新特性,部分函数还没有实现。特别是,目前还不支持新文档的预测。另外也没有Python的API。这些功能后续会添加进来。
 
最大期望[ Expectation Maximization]:
在EM LDA优化器实现分布式LDA模型。
提供的参数LDA:
docConcentration:只有对称先验的支持,所以在提供的k维向量值必须相同,所有的值也必须> 1 > 1.提供向量(- 1)结果默认行为统一k维向量值(50 / K)+ 1
topicConcentration:只有对称的先验的支持。值必须> 1 > 1。在默认值0.1 + 10.1 + 1 - 1提供结果。
maxIterations:EM迭代的最大次数
注:重要的是要做足够的迭代。在早期的迭代,EM经常有无用的主题,但这些主题的显着改善后,更多的迭代。使用至少20个可能50-100迭代往往是合理的,这取决于您的数据集。
EM LDA优化器产生分布式LDA模型,不仅存储推断的主题,但也充分训练语料库和主题分布在训练语料中的每个文件。
topTopicsPerDocument:训练语料库中每个文档的顶级主题及其权重
topDocumentsPerTopic: 每个主题的顶级文档和文档中主题的相应权重。
logPrior:对数概率的估计问题,给出了文档的主题分布的参数各支流和topicconcentration
logLikelihood:训练语料的可能性,给定的推断主题和文档主题分布
 
贝叶斯:
在线LDA优化器和当地的LDA模型的实现。
提供的参数LDA:
docConcentration:非对称信息可以通过在每个k维Dirichlet参数相等的价值载体。值应该> = 0 = 0。提供向量(- 1)结果默认行为(统一k维向量值(1 / K)(1 / K))
topicConcentration:只有对称的先验知识的支持。值必须为= 0 = 0。在默认值为1的结果(1 K)(1 K)。
topicConcentration:只有对称的先验知识的支持。值必须为= 0 = 0。在默认值为1的结果(1 K)(1 K)。
maxIterations: 提交minibatches最大数。
 
在加法,LDA算法在线接受以下参数:
miniBatchFraction: 语料库样本和用于在每一次迭代的部分
optimizeDocConcentration:如果设置为true,执行各支流的超参数的最大似然估计(又名α)在每个minibatch和套在返回的localldamodel优化各支
tau0 and kappa: 用于学习率的衰减,这是由(τ0 +(0 +τITER)?κITER)?κ哪里iteriter是当前迭代次.
 
在线LDA算法产生了一个局部的LDA模型,它只存储推断的主题。一个当地的LDA模型支持:
logLikelihood(documents): 计算给定的给定推断主题的文档的下限.。
logPerplexity(documents): 计算给定的推断的主题提供的文件的困惑的上限。
 
import scala.Tuple2;

import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.clustering.DistributedLDAModel;
import org.apache.spark.mllib.clustering.LDA;
import org.apache.spark.mllib.linalg.Matrix;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.SparkConf;

public class JavaLDAExample {
  public static void main(String[] args) {
    SparkConf conf = new SparkConf().setAppName("LDA Example");
    JavaSparkContext sc = new JavaSparkContext(conf);

    // Load and parse the data
    String path = "data/mllib/sample_lda_data.txt";
    JavaRDD<String> data =http://www.mamicode.com/ sc.textFile(path);
    JavaRDD<Vector> parsedData =http://www.mamicode.com/ data.map(
        new Function<String, Vector>() {
          public Vector call(String s) {
            String[] sarray = s.trim().split(" ");
            double[] values = new double[sarray.length];
            for (int i = 0; i < sarray.length; i++)
              values[i] = Double.parseDouble(sarray[i]);
            return Vectors.dense(values);
          }
        }
    );
    // Index documents with unique IDs
    JavaPairRDD<Long, Vector> corpus = JavaPairRDD.fromJavaRDD(parsedData.zipWithIndex().map(
        new Function<Tuple2<Vector, Long>, Tuple2<Long, Vector>>() {
          public Tuple2<Long, Vector> call(Tuple2<Vector, Long> doc_id) {
            return doc_id.swap();
          }
        }
    ));
    corpus.cache();

    // Cluster the documents into three topics using LDA
    DistributedLDAModel ldaModel = new LDA().setK(3).run(corpus);

    // Output topics. Each is a distribution over words (matching word count vectors)
    System.out.println("Learned topics (as distributions over vocab of " + ldaModel.vocabSize()
        + " words):");
    Matrix topics = ldaModel.topicsMatrix();
    for (int topic = 0; topic < 3; topic++) {
      System.out.print("Topic " + topic + ":");
      for (int word = 0; word < ldaModel.vocabSize(); word++) {
        System.out.print(" " + topics.apply(word, topic));
      }
      System.out.println();
    }

    ldaModel.save(sc.sc(), "myLDAModel");
    DistributedLDAModel sameModel = DistributedLDAModel.load(sc.sc(), "myLDAModel");
  }}

 

二分K均值[Bisecting k-means]:
平分k-均值通常可以比普通的k-均值快得多,但它通常会产生不同的聚类。
Agglomerative:平分k-均值是一种层次聚类。层次聚类是最常用的聚类分析方法之一,旨在建立一个层次结构的集群。层次聚类的策略一般分为两类:
Divisive: 这是一个“自上而下”的方法:所有的观察开始在一个集群,分裂进行递归作为一个向下移动的层次结构。
平分k-均值算法是一种分裂算法。在MLlib的实施具有以下参数:
k:期望的叶子簇数(默认值:4)。实际数字可能没有整除的叶簇较小。
maxIterations: k-均值分裂分裂的最大次数(默认值:20)
minDivisibleClusterSize: 点的最小数目(如果> = 1)或最低比例的点(如<1)的一个分簇(默认值:1)
seed:随机种子(默认值:类名的哈希值)
 
import com.google.common.collect.Lists;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.clustering.BisectingKMeans;
import org.apache.spark.mllib.clustering.BisectingKMeansModel;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;

ArrayList<Vector> localData =http://www.mamicode.com/ Lists.newArrayList(
  Vectors.dense(0.1, 0.1),   Vectors.dense(0.3, 0.3),
  Vectors.dense(10.1, 10.1), Vectors.dense(10.3, 10.3),
  Vectors.dense(20.1, 20.1), Vectors.dense(20.3, 20.3),
  Vectors.dense(30.1, 30.1), Vectors.dense(30.3, 30.3));JavaRDD<Vector> data = http://www.mamicode.com/sc.parallelize(localData, 2);

BisectingKMeans bkm = new BisectingKMeans()
  .setK(4);BisectingKMeansModel model = bkm.run(data);

System.out.println("Compute Cost: " + model.computeCost(data));for (Vector center: model.clusterCenters()) {
  System.out.println("");}Vector[] clusterCenters = model.clusterCenters();for (int i = 0; i < clusterCenters.length; i++) {
  Vector clusterCenter = clusterCenters[i];
  System.out.println("Cluster Center " + i + ": " + clusterCenter);
}
 
流式K均值[Streaming k-means]:
当数据以流式到达,就需要动态预测分类,每当新数据到来时要更新模型。MLlib提供了流式k均值聚类,该方法使用参数来控制数据的衰减。这个算法使用mini-batch k均值更新规则的一种泛化版本。对于每一批数据,将所有点赋给最近的簇,计算新的簇中心,然后使用下面的方法更新簇:
技术分享

 

其中c
衰减可以通过使用halfLife参数指定。对于时刻t取得的数据,在t+halfLife时刻贡献度会降到0.5。
 
在计算是聚类中心的前,nt分配给聚类的点的数目,xt从目前一批新的聚类中心,而mt是点数添加到聚类在当前批处理。衰减因子αα可以忽略过去:与αα= 1的所有数据将被从开始使用;与αα= 0只最新的数据将被使用。这类似于指数加权移动平均。
衰减可以使用半衰期参数指定,确定正确的衰减因子等,在时间t的采集数据,通过时间t +半衰期的贡献将下降到0.5。时间单位可以指定为批次或点和更新规则将相应调整。
 

五、聚类