{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/42821656","id":42821656,"diff_hunk":"@@ -0,0 +1,690 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one or more\n+ * contributor license agreements.  See the NOTICE file distributed with\n+ * this work for additional information regarding copyright ownership.\n+ * The ASF licenses this file to You under the Apache License, Version 2.0\n+ * (the \"License\"); you may not use this file except in compliance with\n+ * the License.  You may obtain a copy of the License at\n+ *\n+ *    http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+package org.apache.spark.mllib.clustering\n+\n+import scala.collection.{Map, mutable}\n+\n+import breeze.linalg\n+  .{SparseVector => BSV, Vector => BV, any => breezeAny, norm => breezeNorm, sum => breezeSum}\n+\n+import org.apache.spark.{Logging, SparkException}\n+import org.apache.spark.annotation.Since\n+import org.apache.spark.mllib.linalg.{Vector, Vectors}\n+import org.apache.spark.rdd.RDD\n+\n+\n+/**\n+ * This is a divisive hierarchical clustering algorithm based on bisecting k-means algorithm.\n+ *\n+ * The main idea of this algorithm is based on \"A comparison of document clustering techniques\",\n+ * M. Steinbach, G. Karypis and V. Kumar. Workshop on Text Mining, KDD, 2000.\n+ * http://cs.fit.edu/~pkc/classes/ml-internet/papers/steinbach00tr.pdf\n+ *\n+ * However, we modified it to fit for Spark. This algorithm consists of the two main parts.\n+ *\n+ * 1. Split clusters until the number of clusters will be enough to build a cluster tree\n+ * 2. Build a cluster tree as a binary tree by the splitted clusters\n+ *\n+ * First, it splits clusters to their children clusters step by step, not considering a cluster\n+ * will be included in the final cluster tree or not. That's because it makes the algorithm more\n+ * efficient on Spark and splitting a cluster one by one is very slow. It will keep splitting until\n+ * the number of clusters will be enough to build a cluster tree. Otherwise, it will stop splitting\n+ * when there are no dividable clusters before the number of clusters will be sufficient. And\n+ * it calculates the criterions, such as average cost, entropy and so on, for building a cluster\n+ * tree in the first part. The criterion means how large the cluster is. That is, the cluster\n+ * whose criterion is maximum of all the clusters is the largest cluster.\n+ *\n+ * Second, it builds a cluster tree as a binary tree by the result of the first part.\n+ * First of all, the cluster tree starts with only the root cluster which includes all points.\n+ * So, there are two candidates which can be merged to the cluster tree. Those are the children of\n+ * the root. Then, it picks up the larger child of the two and merge it to the cluster tree.\n+ * After that, there are tree candidates to merge. Those are the smaller child of the root and\n+ * the two children of the larger cluster of the root. It picks up the largest cluster of the tree\n+ * and merge it to the * cluster tree. Like this, it continues to pick up the largest one of the\n+ * candidates and merge it to the cluster tree until the desired number of clusters is reached.\n+ *\n+ * @param k tne desired number of clusters\n+ * @param clusterMap the pairs of cluster and its index as Map\n+ * @param maxIterations the number of maximal iterations to split clusters\n+ * @param seed a random seed\n+ */\n+@Since(\"1.6.0\")\n+class BisectingKMeans private (\n+    private var k: Int,\n+    private var clusterMap: Map[BigInt, BisectingClusterNode],\n+    private var maxIterations: Int,\n+    private var seed: Long) extends Logging {\n+\n+  /**\n+   * Constructs with the default configuration\n+   */\n+  @Since(\"1.6.0\")\n+  def this() = this(20, mutable.ListMap.empty[BigInt, BisectingClusterNode], 20, 1)\n+\n+  /**\n+   * Sets the number of clusters you want\n+   */\n+  @Since(\"1.6.0\")\n+  def setK(k: Int): this.type = {\n+    this.k = k\n+    this\n+  }\n+\n+  @Since(\"1.6.0\")\n+  def getK: Int = this.k\n+\n+  /**\n+   * Sets the number of maximal iterations in each clustering step\n+   */\n+  @Since(\"1.6.0\")\n+  def setMaxIterations(maxIterations: Int): this.type = {\n+    this.maxIterations = maxIterations\n+    this\n+  }\n+\n+  @Since(\"1.6.0\")\n+  def getMaxIterations: Int = this.maxIterations\n+\n+  /**\n+   * Sets the random seed\n+   */\n+  @Since(\"1.6.0\")\n+  def setSeed(seed: Long): this.type = {\n+    this.seed = seed\n+    this\n+  }\n+\n+  @Since(\"1.6.0\")\n+  def getSeed: Long = this.seed\n+\n+  /**\n+   * Runs the bisecting k-means algorithm\n+   * @param input RDD of vectors\n+   * @return model for the bisecting kmeans\n+   */\n+  @Since(\"1.6.0\")\n+  def run(input: RDD[Vector]): BisectingKMeansModel = {\n+    val sc = input.sparkContext\n+\n+    // `clusterStats` is described as binary tree structure\n+    // `clusterStats(1)` means the root of a binary tree\n+    var clusterStats = mutable.Map.empty[BigInt, BisectingClusterStat]\n+    var step = 1\n+    var noMoreDividable = false\n+    var rddArray = Array.empty[RDD[(BigInt, BV[Double])]]\n+    // the number of maximum nodes of a binary tree by given parameter\n+    val multiplier = math.ceil(math.log10(this.k) / math.log10(2.0)) + 1\n+    val maxAllNodesInTree = math.pow(2, multiplier).toInt\n+\n+    // divide clusters until the number of clusters reachs the condition\n+    // or there is no dividable cluster\n+    val startTime = System.currentTimeMillis()\n+    var data = BisectingKMeans.initData(input).cache()\n+    while (clusterStats.size < maxAllNodesInTree && noMoreDividable == false) {\n+      logInfo(s\"${sc.appName} starts step ${step}\")\n+      val leafClusters = BisectingKMeans.summarizeClusters(data)\n+      val dividableLeafClusters = leafClusters.filter(_._2.isDividable)\n+      clusterStats = clusterStats ++ leafClusters\n+\n+      if (dividableLeafClusters.isEmpty) {\n+        noMoreDividable = true\n+      }\n+      else {\n+        // can be clustered if the number of divided clusterStats is equal to 0\n+        val divided =\n+          BisectingKMeans.divideClusters(data, dividableLeafClusters, maxIterations)\n+        // update each index\n+        val newData = BisectingKMeans.updateClusterIndex(data, divided).cache()\n+        rddArray = rddArray ++ Array(data)\n+        data = newData\n+        // keep recent 2 cached RDDs in order to run more quickly\n+        if (rddArray.length > 1) {\n+          val head = rddArray.head\n+          head.unpersist()\n+          rddArray = rddArray.filterNot(_.hashCode() == head.hashCode())\n+        }\n+        clusterStats = clusterStats ++ divided\n+        step += 1\n+        logInfo(s\"${sc.appName} adding ${divided.size} new clusterStats at step:${step}\")\n+      }\n+    }\n+    // unpersist kept RDDs\n+    rddArray.foreach(_.unpersist())\n+    // create a map of cluster node with their criterions\n+    val nodes = BisectingKMeans.createClusterNodes(data, clusterStats)\n+\n+    // build a cluster tree by Map class which is expressed\n+    logInfo(s\"Building the cluster tree is started in ${sc.appName}\")\n+    val root = BisectingKMeans.buildTree(nodes, BisectingKMeans.ROOT_INDEX_KEY, this.k)\n+    if (root.isEmpty) {\n+      new SparkException(\"Failed to build a cluster tree from a Map type of clusterStats\")\n+    }\n+\n+    // set the elapsed time for training\n+    val finishTime = (System.currentTimeMillis() - startTime) / 1000.0\n+    logInfo(s\"Elapsed Time for ${this.getClass.getSimpleName} Training: ${finishTime} [sec]\")\n+\n+    // make a bisecting kmeans model\n+    val model = new BisectingKMeansModel(root.get)\n+    val leavesNodes = model.getClusters\n+    if (leavesNodes.length < this.k) {\n+      logWarning(s\"# clusters is less than you want: ${leavesNodes.length} / ${k}\")\n+    }\n+    model\n+  }\n+}\n+\n+\n+private[clustering] object BisectingKMeans {\n+\n+  val ROOT_INDEX_KEY: BigInt = 1\n+\n+  /**\n+   * Finds the closes cluster's center\n+   *\n+   * @param metric a distance metric\n+   * @param centers centers of the clusters\n+   * @param point a target point\n+   * @return an index of the array of clusters\n+   */\n+  def findClosestCenter(metric: (BV[Double], BV[Double]) => Double)\n+      (centers: Seq[BV[Double]])(point: BV[Double]): Int = {\n+    val (closestCenter, closestIndex) =\n+      centers.zipWithIndex.map { case (center, idx) => (metric(center, point), idx)}.minBy(_._1)\n+    closestIndex\n+  }\n+\n+  /**\n+   * Summarizes data by each cluster as Map\n+   *\n+   * @param data pairs of point and its cluster index\n+   */\n+  def summarizeClusters(data: RDD[(BigInt, BV[Double])]): Map[BigInt, BisectingClusterStat] = {\n+\n+    val stats = data.mapPartitions { iter =>\n+      // calculate the accumulation of the all point in a partition and count the rows\n+      val map = mutable.Map.empty[BigInt, (BV[Double], Double, BV[Double])]\n+      iter.foreach { case (idx: BigInt, point: BV[Double]) =>\n+        // get a map value or else get a sparse vector\n+        val (sumBV, n, sumOfSquares) = map\n+          .getOrElse(idx, (BSV.zeros[Double](point.size), 0.0, BSV.zeros[Double](point.size)))\n+        map(idx) = (sumBV + point, n + 1.0, sumOfSquares + (point :* point))\n+      }\n+      map.toIterator\n+    }.reduceByKey { case ((sum1, n1, sumOfSquares1), (sum2, n2, sumOfSquares2)) =>\n+      // sum the accumulation and the count in the all partition\n+      (sum1 + sum2, n1 + n2, sumOfSquares1 + sumOfSquares2)\n+    }.collect().toMap\n+\n+    stats.map {case (i, stat) => i -> new BisectingClusterStat(stat._2.toLong, stat._1, stat._3)}\n+  }\n+\n+  /**\n+   * Assigns the initial cluster index id to all data\n+   */\n+  def initData(data: RDD[Vector]): RDD[(BigInt, BV[Double])] = {\n+    data.map { v: Vector => (BisectingKMeans.ROOT_INDEX_KEY, v.toBreeze)}\n+  }\n+\n+  /**\n+   * Gets the initial centers for bisect k-means\n+   *\n+   * @param data pairs of point and its cluster index\n+   * @param stats pairs of cluster index and cluster statistics\n+   */\n+  def initNextCenters(\n+      data: RDD[(BigInt, BV[Double])],\n+      stats: Map[BigInt, BisectingClusterStat]): Map[BigInt, BV[Double]] = {\n+\n+    // Since the combination sampleByKey and groupByKey is more expensive,\n+    // this as follows would be better.\n+    val bcIndeces = data.sparkContext.broadcast(stats.keySet)\n+    val samples = data.mapPartitions { iter =>\n+      val map = mutable.Map.empty[BigInt, mutable.ArrayBuffer[BV[Double]]]\n+\n+      bcIndeces.value.foreach {i => map(i) = mutable.ArrayBuffer.empty[BV[Double]]}\n+      val LOCAL_SAMPLE_SIZE = 100\n+      iter.foreach { case (i, point) =>\n+        map(i).append(point)\n+        // to avoid to increase the memory usage on each map thread,\n+        // the number of elements is cut off at the right time.\n+        if (map(i).size > LOCAL_SAMPLE_SIZE) {\n+          val elements = map(i).sortWith((a, b) => breezeNorm(a, 2.0) < breezeNorm(b, 2.0))\n+          map(i) = mutable.ArrayBuffer(elements.head, elements.last)\n+        }\n+      }\n+\n+      // in order to reduce the shuffle size, take only two elements\n+      map.filterNot(_._2.isEmpty).map { case (i, points) =>\n+        val elements = map(i).toSeq.sortWith((a, b) => breezeNorm(a, 2.0) < breezeNorm(b, 2.0))\n+        i -> mutable.ArrayBuffer(elements.head, elements.last)\n+      }.toIterator\n+    }.reduceByKey { case (points1, points2) =>\n+      points1.union(points2)\n+    }.collect()\n+\n+    val nextCenters = samples.flatMap { case (i, points) =>\n+      val elements = points.toSeq.sortWith((a, b) => breezeNorm(a, 2.0) < breezeNorm(b, 2.0))\n+      Array((2 * i, elements.head), (2 * i + 1, elements.last))\n+    }.toMap\n+    if (!stats.keySet.flatMap(idx => Array(2 * idx, 2 * idx + 1)).forall(nextCenters.contains(_))) {\n+      throw new SparkException(\"Failed to initialize centers for next step\")\n+    }\n+    nextCenters\n+  }\n+\n+  /**\n+   * Updates the indexes of clusters which is divided to its children indexes\n+   *\n+   * @param data pairs of point and its cluster index\n+   * @param dividedClusters pairs of cluster index and cluster statistics\n+   */\n+  def updateClusterIndex(\n+      data: RDD[(BigInt, BV[Double])],\n+      dividedClusters: Map[BigInt, BisectingClusterStat]): RDD[(BigInt, BV[Double])] = {\n+\n+    // extract the centers of the clusters\n+    val sc = data.sparkContext\n+    var centers = dividedClusters.map { case (idx, cluster) => (idx, cluster.center)}\n+    val bcCenters = sc.broadcast(centers)\n+\n+    // TODO Supports distance metrics other Euclidean distance metric\n+    val metric = (bv1: BV[Double], bv2: BV[Double]) => breezeNorm(bv1 - bv2, 2.0)\n+    val bcMetric = sc.broadcast(metric)\n+\n+    // update the indexes to their children indexes\n+    data.map { case (idx, point) =>\n+      val childrenIndexes = Array(2 * idx, 2 * idx + 1).filter(c => bcCenters.value.contains(c))\n+      childrenIndexes.length match {\n+        // stay the index if the number of children is not enough\n+        case s if s < 2 => (idx, point)\n+        // update the indexes\n+        case _ => {\n+          val nextCenters = childrenIndexes.map(bcCenters.value(_))\n+          val closestIndex = BisectingKMeans\n+            .findClosestCenter(bcMetric.value)(nextCenters)(point)\n+          val nextIndex = 2 * idx + closestIndex\n+          (nextIndex, point)\n+        }\n+      }\n+    }\n+  }\n+\n+  /**\n+   * Divides clusters according to their statistics\n+   *\n+   * @param data pairs of point and its cluster index\n+   * @param targetClusters target clusters to divide\n+   * @param maxIterations the maximum iterations to calculate clusters statistics\n+   */\n+  def divideClusters(\n+      data: RDD[(BigInt, BV[Double])],\n+      targetClusters: Map[BigInt, BisectingClusterStat],\n+      maxIterations: Int): Map[BigInt, BisectingClusterStat] = {\n+    val sc = data.sparkContext\n+    val appName = sc.appName\n+\n+    // get keys of dividable clusters\n+    val dividableClusters = targetClusters.filter { case (idx, cluster) => cluster.isDividable }\n+    if (dividableClusters.isEmpty) {\n+      return Map.empty[BigInt, BisectingClusterStat]\n+    }\n+    // extract dividable input data\n+    val dividableData = data.filter { case (idx, point) => dividableClusters.contains(idx)}\n+\n+    var newCenters = BisectingKMeans.initNextCenters(dividableData, dividableClusters)\n+    var bcNewCenters = sc.broadcast(newCenters)\n+    // TODO Supports distance metrics other Euclidean distance metric\n+    val metric = (bv1: BV[Double], bv2: BV[Double]) => breezeNorm(bv1 - bv2, 2.0)\n+    val bcMetric = sc.broadcast(metric)\n+    var stats = Map.empty[BigInt, (BV[Double], Double, BV[Double])]\n+\n+    var subIter = 0\n+    var totalStd = Double.MaxValue\n+    var oldTotalStd = Double.MaxValue\n+    var relativeError = Double.MaxValue\n+    while (subIter < maxIterations && relativeError > 10E-4) {\n+      // calculate summary of each cluster\n+      val eachStats = dividableData.mapPartitions { iter =>\n+        val map = mutable.Map.empty[BigInt, (BV[Double], Double, BV[Double])]\n+        iter.foreach { case (idx, point) =>\n+          // calculate next index number\n+          val childrenCenters = Array(2 * idx, 2 * idx + 1)\n+            .filter(x => bcNewCenters.value.contains(x)).map(bcNewCenters.value(_))\n+          if (childrenCenters.length == 2) {\n+            val closestIndex =\n+              BisectingKMeans.findClosestCenter(bcMetric.value)(childrenCenters)(point)\n+            val nextIndex = 2 * idx + closestIndex\n+\n+            // get a map value or else get a sparse vector\n+            val (sumBV, n, sumOfSquares) = map\n+              .getOrElse(\n+                nextIndex,\n+                (BSV.zeros[Double](point.size), 0.0, BSV.zeros[Double](point.size))\n+              )\n+            map(nextIndex) = (sumBV + point, n + 1.0, sumOfSquares + (point :* point))\n+          }\n+        }\n+        map.toIterator\n+      }.reduceByKey { case ((sv1, n1, sumOfSquares1), (sv2, n2, sumOfSquares2)) =>\n+        // sum the accumulation and the count in the all partition\n+        (sv1 + sv2, n1 + n2, sumOfSquares1 + sumOfSquares2)\n+      }.collect().toMap\n+\n+      // calculate the center of each cluster\n+      newCenters = eachStats.map { case (idx, (sum, n, sumOfSquares)) => (idx, sum :/ n)}\n+      bcNewCenters = sc.broadcast(newCenters)\n+\n+      // update summary of each cluster\n+      stats = eachStats.toMap\n+\n+      totalStd = stats.map { case (idx, (sum, n, sumOfSquares)) =>\n+        breezeSum((sumOfSquares :/ n) :- breezeNorm(sum :/ n, 2.0))\n+      }.sum\n+      relativeError = math.abs(oldTotalStd - totalStd) / totalStd\n+      oldTotalStd = totalStd\n+      subIter += 1\n+    }\n+    stats.map { case (i, stat) => i -> new BisectingClusterStat(stat._2.toLong, stat._1, stat._3) }\n+  }\n+\n+  /**\n+   * Creates the map of cluster stats to the map of cluster nodes with their criterions\n+   *\n+   * @param data input data\n+   * @param stats map of cluster stats which is described as a binary tree\n+   */\n+  def createClusterNodes(\n+      data: RDD[(BigInt, BV[Double])],\n+      stats: Map[BigInt, BisectingClusterStat]): Map[BigInt, BisectingClusterNode] = {\n+\n+    // TODO: support other criterion, such as entropy\n+    createClusterNodesWithAverageCost(data, stats)\n+  }\n+\n+  /**\n+   * Creates the map of cluster stats to the map of cluster nodes with their average costs\n+   */\n+  private def createClusterNodesWithAverageCost(\n+      data: RDD[(BigInt, BV[Double])],\n+      stats: Map[BigInt, BisectingClusterStat]): Map[BigInt, BisectingClusterNode] = {\n+\n+    // calculate average costs of all clusters\n+    val bcCenters = data.sparkContext.broadcast(stats.map { case (i, stat) => i -> stat.center })\n+    val costs = data.mapPartitions { iter =>\n+      val counters = mutable.Map.empty[BigInt, (Long, Double)]\n+      bcCenters.value.foreach {case (i, center) => counters(i) = (0L, 0.0)}\n+      iter.foreach { case (i, point) =>\n+        val cost = breezeNorm(bcCenters.value.apply(i) - point, 2.0)\n+        counters(i) = (counters(i)._1 + 1, counters(i)._2 + cost)\n+      }\n+      counters.toIterator\n+    }.reduceByKey { case((n1, cost1), (n2, cost2)) =>\n+      (n1 + n2, cost1 + cost2)\n+    }.collectAsMap()\n+\n+    stats.map { case (i, stat) =>\n+      val avgCost = costs(i)._1 match {\n+        case x if x == 0.0 => 0.0\n+        case _ => costs(i)._2 / costs(i)._1\n+      }\n+      i -> new BisectingClusterNode(Vectors.fromBreeze(stat.center), stat.rows, avgCost)\n+    }\n+  }\n+\n+  /**\n+   * Builds a cluster tree from a Map of clusters\n+   *\n+   * @param treeMap divided clusters as a Map class\n+   * @param rootIndex index you want to start\n+   * @param numClusters the number of clusters you want\n+   * @return a built cluster tree\n+   */\n+  private def buildTree(\n+      treeMap: Map[BigInt, BisectingClusterNode],\n+      rootIndex: BigInt,\n+      numClusters: Int): Option[BisectingClusterNode] = {\n+\n+    // if there is no index in the Map\n+    if (!treeMap.contains(rootIndex)) return None\n+\n+    // build a cluster tree if the queue is empty or until the number of leaf clusters is enough\n+    var numLeavesClusters = 1\n+    val root = treeMap(rootIndex)\n+    var leavesQueue = Map(rootIndex -> root)\n+    while (leavesQueue.nonEmpty && numLeavesClusters < numClusters) {\n+      // pick up the largest cluster by the maximum criterion of all the clusters\n+      val mostScattered = leavesQueue.maxBy(_._2.criterion)\n+      val mostScatteredKey = mostScattered._1\n+      val mostScatteredCluster = mostScattered._2\n+\n+      // relate the most scattered cluster to its children clusters\n+      val childrenIndexes = Array(2 * mostScatteredKey, 2 * mostScatteredKey + 1)\n+      if (childrenIndexes.forall(i => treeMap.contains(i))) {\n+        // insert children to the most scattered cluster\n+        val children = childrenIndexes.map(i => treeMap(i))\n+        mostScatteredCluster.insert(children)\n+\n+        // calculate the local dendrogram height\n+        // TODO Supports distance metrics other Euclidean distance metric\n+        val metric = (bv1: BV[Double], bv2: BV[Double]) => breezeNorm(bv1 - bv2, 2.0)\n+        val localHeight = children\n+          .map(child => metric(child.center.toBreeze, mostScatteredCluster.center.toBreeze)).max\n+        mostScatteredCluster.setLocalHeight(localHeight)\n+\n+        // update the queue\n+        leavesQueue = leavesQueue ++ childrenIndexes.map(i => i -> treeMap(i)).toMap\n+        numLeavesClusters += 1\n+      }\n+\n+      // remove the cluster which is involved to the cluster tree\n+      leavesQueue = leavesQueue.filterNot(_ == mostScattered)\n+    }\n+    Some(root)\n+  }\n+}\n+\n+/**\n+ * A cluster as a tree node which can have its sub nodes\n+ *\n+ * @param center the center of the cluster\n+ * @param rows the number of rows in the cluster\n+ * @param criterion how large a cluster is\n+ * @param localHeight the maximal distance between this node and its children\n+ * @param parent the parent cluster of the cluster\n+ * @param children the children nodes of the cluster\n+ */\n+@Since(\"1.6.0\")\n+class BisectingClusterNode private (\n+    @Since(\"1.6.0\") val center: Vector,\n+    @Since(\"1.6.0\") val rows: Long,\n+    @Since(\"1.6.0\") val criterion: Double,\n+    private var localHeight: Double,\n+    private var parent: Option[BisectingClusterNode],\n+    private var children: Seq[BisectingClusterNode]) extends Serializable {\n+\n+  require(!criterion.isNaN)\n+\n+  @Since(\"1.6.0\")\n+  def this(center: Vector, rows: Long, criterion: Double) =\n+    this(center, rows, criterion, 0.0, None, Array.empty[BisectingClusterNode])\n+\n+  /**\n+   * Inserts a sub node as its child\n+   *\n+   * @param child inserted sub node\n+   */\n+  @Since(\"1.6.0\")\n+  def insert(child: BisectingClusterNode) {\n+    insert(Array(child))\n+  }\n+\n+  /**\n+   * Inserts sub nodes as its children\n+   *\n+   * @param children inserted sub nodes\n+   */\n+  @Since(\"1.6.0\")\n+  def insert(children: Array[BisectingClusterNode]) {\n+    this.children = this.children ++ children\n+    children.foreach(child => child.parent = Some(this))\n+  }\n+\n+  /**\n+   * Converts the tree into Array class\n+   * the sub nodes are recursively expanded\n+   *\n+   * @return an Array class which the cluster tree is expanded\n+   */\n+  @Since(\"1.6.0\")\n+  def toArray: Array[BisectingClusterNode] = {\n+    val array = this.children.size match {\n+      case 0 => Array(this)\n+      case _ => Array(this) ++ this.children.flatMap(child => child.toArray.toIterator)\n+    }\n+    array.sortWith { case (a, b) =>\n+      a.getDepth < b.getDepth && a.criterion < b.criterion && a.rows < b.rows\n+    }\n+  }\n+\n+  /**\n+   * Gets the depth of the cluster in the tree\n+   *\n+   * @return the depth from the root\n+   */\n+  @Since(\"1.6.0\")\n+  def getDepth: Int = {\n+    this.parent match {\n+      case None => 0\n+      case _ => 1 + this.parent.get.getDepth\n+    }\n+  }\n+\n+  /**\n+   * Gets the leaves nodes in the cluster tree\n+   */\n+  @Since(\"1.6.0\")\n+  def getLeavesNodes: Array[BisectingClusterNode] = {\n+    this.toArray.filter(_.isLeaf).sortBy(_.center.toArray.sum)\n+  }\n+\n+  @Since(\"1.6.0\")\n+  def isLeaf: Boolean = this.children.isEmpty\n+\n+  @Since(\"1.6.0\")\n+  def getParent: Option[BisectingClusterNode] = this.parent\n+\n+  @Since(\"1.6.0\")\n+  def getChildren: Seq[BisectingClusterNode] = this.children\n+\n+  /**\n+   * Gets the dendrogram height of the cluster at the cluster tree.\n+   * A dendrogram height is different from a local height.\n+   * A dendrogram height means a total height of a node in a tree.\n+   * A local height means a maximum distance between a node and its children.\n+   *\n+   * @return the dendrogram height\n+   */\n+  @Since(\"1.6.0\")\n+  def getHeight: Double = {\n+    this.children.size match {\n+      case 0 => 0.0\n+      case _ => this.localHeight + this.children.map(_.getHeight).max\n+    }\n+  }\n+\n+  @Since(\"1.6.0\")\n+  def setLocalHeight(height: Double): Unit = this.localHeight = height\n+\n+  /**\n+   * Converts to an adjacency list\n+   *\n+   * @return List[(fromNodeId, toNodeId, distance)]\n+   */\n+  @Since(\"1.6.0\")\n+  def toAdjacencyList: Array[(Int, Int, Double)] = {\n+    val nodes = toArray\n+\n+    var adjacencyList = Array.empty[(Int, Int, Double)]\n+    nodes.foreach { parent =>\n+      if (parent.children.size > 1) {\n+        val parentIndex = nodes.indexOf(parent)\n+        parent.children.foreach { child =>\n+          val childIndex = nodes.indexOf(child)\n+          adjacencyList = adjacencyList :+(parentIndex, childIndex, parent.localHeight)\n+        }\n+      }\n+    }\n+    adjacencyList\n+  }\n+\n+  /**\n+   * Converts to a linkage matrix\n+   * Returned data format is fit for scipy's dendrogram function\n+   *\n+   * @return List[(node1, node2, distance, tree size)]\n+   */\n+  @Since(\"1.6.0\")\n+  def toLinkageMatrix: Array[(Int, Int, Double, Int)] = {\n+    val nodes = toArray.sortWith { case (a, b) => a.getHeight < b.getHeight}\n+    val leaves = nodes.filter(_.isLeaf)\n+    val notLeaves = nodes.filterNot(_.isLeaf).filter(_.getChildren.size > 1)\n+    val clusters = leaves ++ notLeaves\n+    val treeMap = clusters.zipWithIndex.map { case (node, idx) => node -> idx}.toMap\n+\n+    // If a node only has one-child, the child is regarded as the cluster of the child.\n+    // Cluster A has cluster B and Cluster B. B is a leaf. C only has cluster D.\n+    // ==> A merge list is (B, D), not (B, C).\n+    def getIndex(map: Map[BisectingClusterNode, Int], node: BisectingClusterNode): Int = {\n+      node.children.size match {\n+        case 1 => getIndex(map, node.children.head)\n+        case _ => map(node)\n+      }\n+    }\n+    clusters.filterNot(_.isLeaf).map { node =>\n+      (getIndex(treeMap, node.children.head),\n+        getIndex(treeMap, node.children(1)),\n+        node.getHeight,\n+        node.toArray.filter(_.isLeaf).length)\n+    }\n+  }\n+}\n+\n+\n+/**\n+ *  This class is used for maneging a cluster statistics\n+ *\n+ * @param rows the number of points\n+ * @param sums the sum of points\n+ * @param sumOfSquares the sum of squares of points\n+ */\n+private[clustering] case class BisectingClusterStat (\n+    rows: Long,\n+    sums: BV[Double],\n+    sumOfSquares: BV[Double]) extends Serializable {\n+\n+  // initialization\n+  val center: BV[Double] = sums :/ rows.toDouble\n+  val variances: BV[Double] = rows match {","path":"mllib/src/main/scala/org/apache/spark/mllib/clustering/BisectingKMeans.scala","position":683,"original_position":683,"commit_id":"31623ead10d4664c7c5d98b5c72731bcc804bcba","original_commit_id":"31623ead10d4664c7c5d98b5c72731bcc804bcba","user":{"login":"mengxr","id":829644,"avatar_url":"https://avatars.githubusercontent.com/u/829644?v=3","gravatar_id":"","url":"https://api.github.com/users/mengxr","html_url":"https://github.com/mengxr","followers_url":"https://api.github.com/users/mengxr/followers","following_url":"https://api.github.com/users/mengxr/following{/other_user}","gists_url":"https://api.github.com/users/mengxr/gists{/gist_id}","starred_url":"https://api.github.com/users/mengxr/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mengxr/subscriptions","organizations_url":"https://api.github.com/users/mengxr/orgs","repos_url":"https://api.github.com/users/mengxr/repos","events_url":"https://api.github.com/users/mengxr/events{/privacy}","received_events_url":"https://api.github.com/users/mengxr/received_events","type":"User","site_admin":false},"body":"* use a scalar variance instead of coordinate variance\r\n* change `sums` and `sumOfSquares` to `mean: BV[Double]` and `variance: Double`","created_at":"2015-10-23T00:08:34Z","updated_at":"2015-10-23T00:08:34Z","html_url":"https://github.com/apache/spark/pull/5267#discussion_r42821656","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/5267","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/42821656"},"html":{"href":"https://github.com/apache/spark/pull/5267#discussion_r42821656"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/5267"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/5267","id":32226987,"html_url":"https://github.com/apache/spark/pull/5267","diff_url":"https://github.com/apache/spark/pull/5267.diff","patch_url":"https://github.com/apache/spark/pull/5267.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/5267","number":5267,"state":"open","locked":false,"title":"[SPARK-6517][mllib] Implement the Algorithm of Hierarchical Clustering","user":{"login":"yu-iskw","id":1523515,"avatar_url":"https://avatars.githubusercontent.com/u/1523515?v=3","gravatar_id":"","url":"https://api.github.com/users/yu-iskw","html_url":"https://github.com/yu-iskw","followers_url":"https://api.github.com/users/yu-iskw/followers","following_url":"https://api.github.com/users/yu-iskw/following{/other_user}","gists_url":"https://api.github.com/users/yu-iskw/gists{/gist_id}","starred_url":"https://api.github.com/users/yu-iskw/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/yu-iskw/subscriptions","organizations_url":"https://api.github.com/users/yu-iskw/orgs","repos_url":"https://api.github.com/users/yu-iskw/repos","events_url":"https://api.github.com/users/yu-iskw/events{/privacy}","received_events_url":"https://api.github.com/users/yu-iskw/received_events","type":"User","site_admin":false},"body":"I implemented a hierarchical clustering algorithm again.  This PR doesn't include examples, documentation and spark.ml APIs. I am going to send another PRs later.\r\nhttps://issues.apache.org/jira/browse/SPARK-6517\r\n\r\n- This implementation based on a bi-sectiong K-means clustering.\r\n    - It derives from the @freeman-lab 's implementation\r\n- The basic idea is not changed from the previous version. (#2906)\r\n    - However, It is 1000x faster than the previous version through parallel processing.\r\n\r\nThank you for your great cooperation, RJ Nowling(@rnowling), Jeremy Freeman(@freeman-lab), Xiangrui Meng(@mengxr) and Sean Owen(@srowen).","created_at":"2015-03-30T13:40:12Z","updated_at":"2015-10-23T00:08:34Z","closed_at":null,"merged_at":null,"merge_commit_sha":"5009b373f5f642811f94c490484d9e4b6ebb1c0e","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/5267/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/5267/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/5267/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/31623ead10d4664c7c5d98b5c72731bcc804bcba","head":{"label":"yu-iskw:new-hierarchical-clustering","ref":"new-hierarchical-clustering","sha":"31623ead10d4664c7c5d98b5c72731bcc804bcba","user":{"login":"yu-iskw","id":1523515,"avatar_url":"https://avatars.githubusercontent.com/u/1523515?v=3","gravatar_id":"","url":"https://api.github.com/users/yu-iskw","html_url":"https://github.com/yu-iskw","followers_url":"https://api.github.com/users/yu-iskw/followers","following_url":"https://api.github.com/users/yu-iskw/following{/other_user}","gists_url":"https://api.github.com/users/yu-iskw/gists{/gist_id}","starred_url":"https://api.github.com/users/yu-iskw/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/yu-iskw/subscriptions","organizations_url":"https://api.github.com/users/yu-iskw/orgs","repos_url":"https://api.github.com/users/yu-iskw/repos","events_url":"https://api.github.com/users/yu-iskw/events{/privacy}","received_events_url":"https://api.github.com/users/yu-iskw/received_events","type":"User","site_admin":false},"repo":{"id":20396045,"name":"spark","full_name":"yu-iskw/spark","owner":{"login":"yu-iskw","id":1523515,"avatar_url":"https://avatars.githubusercontent.com/u/1523515?v=3","gravatar_id":"","url":"https://api.github.com/users/yu-iskw","html_url":"https://github.com/yu-iskw","followers_url":"https://api.github.com/users/yu-iskw/followers","following_url":"https://api.github.com/users/yu-iskw/following{/other_user}","gists_url":"https://api.github.com/users/yu-iskw/gists{/gist_id}","starred_url":"https://api.github.com/users/yu-iskw/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/yu-iskw/subscriptions","organizations_url":"https://api.github.com/users/yu-iskw/orgs","repos_url":"https://api.github.com/users/yu-iskw/repos","events_url":"https://api.github.com/users/yu-iskw/events{/privacy}","received_events_url":"https://api.github.com/users/yu-iskw/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/yu-iskw/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/yu-iskw/spark","forks_url":"https://api.github.com/repos/yu-iskw/spark/forks","keys_url":"https://api.github.com/repos/yu-iskw/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/yu-iskw/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/yu-iskw/spark/teams","hooks_url":"https://api.github.com/repos/yu-iskw/spark/hooks","issue_events_url":"https://api.github.com/repos/yu-iskw/spark/issues/events{/number}","events_url":"https://api.github.com/repos/yu-iskw/spark/events","assignees_url":"https://api.github.com/repos/yu-iskw/spark/assignees{/user}","branches_url":"https://api.github.com/repos/yu-iskw/spark/branches{/branch}","tags_url":"https://api.github.com/repos/yu-iskw/spark/tags","blobs_url":"https://api.github.com/repos/yu-iskw/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/yu-iskw/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/yu-iskw/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/yu-iskw/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/yu-iskw/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/yu-iskw/spark/languages","stargazers_url":"https://api.github.com/repos/yu-iskw/spark/stargazers","contributors_url":"https://api.github.com/repos/yu-iskw/spark/contributors","subscribers_url":"https://api.github.com/repos/yu-iskw/spark/subscribers","subscription_url":"https://api.github.com/repos/yu-iskw/spark/subscription","commits_url":"https://api.github.com/repos/yu-iskw/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/yu-iskw/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/yu-iskw/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/yu-iskw/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/yu-iskw/spark/contents/{+path}","compare_url":"https://api.github.com/repos/yu-iskw/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/yu-iskw/spark/merges","archive_url":"https://api.github.com/repos/yu-iskw/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/yu-iskw/spark/downloads","issues_url":"https://api.github.com/repos/yu-iskw/spark/issues{/number}","pulls_url":"https://api.github.com/repos/yu-iskw/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/yu-iskw/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/yu-iskw/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/yu-iskw/spark/labels{/name}","releases_url":"https://api.github.com/repos/yu-iskw/spark/releases{/id}","created_at":"2014-06-02T07:24:57Z","updated_at":"2014-07-25T08:42:00Z","pushed_at":"2015-10-22T22:38:46Z","git_url":"git://github.com/yu-iskw/spark.git","ssh_url":"git@github.com:yu-iskw/spark.git","clone_url":"https://github.com/yu-iskw/spark.git","svn_url":"https://github.com/yu-iskw/spark","homepage":null,"size":121303,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":2,"forks":0,"open_issues":2,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"1d9733271595596683a6d956a7433fa601df1cc1","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":17165658,"name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/spark","description":"Mirror of Apache Spark","fork":false,"url":"https://api.github.com/repos/apache/spark","forks_url":"https://api.github.com/repos/apache/spark/forks","keys_url":"https://api.github.com/repos/apache/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/spark/teams","hooks_url":"https://api.github.com/repos/apache/spark/hooks","issue_events_url":"https://api.github.com/repos/apache/spark/issues/events{/number}","events_url":"https://api.github.com/repos/apache/spark/events","assignees_url":"https://api.github.com/repos/apache/spark/assignees{/user}","branches_url":"https://api.github.com/repos/apache/spark/branches{/branch}","tags_url":"https://api.github.com/repos/apache/spark/tags","blobs_url":"https://api.github.com/repos/apache/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/spark/languages","stargazers_url":"https://api.github.com/repos/apache/spark/stargazers","contributors_url":"https://api.github.com/repos/apache/spark/contributors","subscribers_url":"https://api.github.com/repos/apache/spark/subscribers","subscription_url":"https://api.github.com/repos/apache/spark/subscription","commits_url":"https://api.github.com/repos/apache/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/spark/contents/{+path}","compare_url":"https://api.github.com/repos/apache/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/spark/merges","archive_url":"https://api.github.com/repos/apache/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/spark/downloads","issues_url":"https://api.github.com/repos/apache/spark/issues{/number}","pulls_url":"https://api.github.com/repos/apache/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/spark/labels{/name}","releases_url":"https://api.github.com/repos/apache/spark/releases{/id}","created_at":"2014-02-25T08:00:08Z","updated_at":"2015-10-22T20:01:53Z","pushed_at":"2015-10-22T23:45:30Z","git_url":"git://github.com/apache/spark.git","ssh_url":"git@github.com:apache/spark.git","clone_url":"https://github.com/apache/spark.git","svn_url":"https://github.com/apache/spark","homepage":null,"size":1639468,"stargazers_count":5844,"watchers_count":5844,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":5457,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":333,"forks":5457,"open_issues":333,"watchers":5844,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/5267"},"html":{"href":"https://github.com/apache/spark/pull/5267"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/5267"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/5267/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/5267/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/5267/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/31623ead10d4664c7c5d98b5c72731bcc804bcba"}}}}