You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by srowen <gi...@git.apache.org> on 2016/10/04 10:26:49 UTC
[GitHub] spark pull request #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMean...
GitHub user srowen opened a pull request:
https://github.com/apache/spark/pull/15342
[SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans implementation / remove 'runs' / KMeans clusterer can return duplicate cluster centers
## What changes were proposed in this pull request?
This is a revival of https://github.com/apache/spark/pull/14948 and related to https://github.com/apache/spark/pull/14937. This removes the 'runs' parameter, which has already been disabled, from the K-means implementation and further deprecates API methods that involve it.
This also happens to resolve the issue that K-means should not return duplicate centers, meaning that it may return less than k centroids if not enough data is available.
## How was this patch tested?
Existing tests
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/srowen/spark SPARK-11560
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/15342.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #15342
----
commit cd14b65ac082d728d3a99aeede54c9a7e1f22614
Author: Sean Owen <so...@cloudera.com>
Date: 2016-10-04T10:24:03Z
Remove inactive KMeans 'runs' param; allow fewer than k centroids to be chosen too
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82649893
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
+ totalContribs.foreach { case (j, (sum, count)) =>
+ scal(1.0 / count, sum)
+ val newCenter = new VectorWithNorm(sum)
+ if (converged && KMeans.fastSquaredDistance(newCenter, centers(j)) > epsilon * epsilon) {
+ converged = false
}
- costs(run) = costAccums(i).value
+ centers(j) = newCenter
}
- activeRuns = activeRuns.filter(active(_))
+ cost = costAccum.value
iteration += 1
}
val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
- logInfo(s"Iterations took " + "%.3f".format(iterationTimeInSeconds) + " seconds.")
+ logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")
if (iteration == maxIterations) {
logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
} else {
logInfo(s"KMeans converged in $iteration iterations.")
}
- val (minCost, bestRun) = costs.zipWithIndex.min
+ logInfo(s"The cost is $cost.")
- logInfo(s"The cost for the best run is $minCost.")
-
- new KMeansModel(centers(bestRun).map(_.vector))
+ new KMeansModel(centers.map(_.vector))
}
/**
- * Initialize `runs` sets of cluster centers at random.
+ * Initialize a set of cluster centers at random.
*/
- private def initRandom(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
- // Sample all the cluster centers in one pass to avoid repeated scans
- val sample = data.takeSample(true, runs * k, new XORShiftRandom(this.seed).nextInt()).toSeq
- Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
- new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm)
- }.toArray)
+ private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
+ data.takeSample(true, k, new XORShiftRandom(this.seed).nextInt()).map(_.toDense)
}
/**
- * Initialize `runs` sets of cluster centers using the k-means|| algorithm by Bahmani et al.
+ * Initialize a set of cluster centers using the k-means|| algorithm by Bahmani et al.
* (Bahmani et al., Scalable K-Means++, VLDB 2012). This is a variant of k-means++ that tries
- * to find with dissimilar cluster centers by starting with a random center and then doing
+ * to find dissimilar cluster centers by starting with a random center and then doing
* passes where more centers are chosen with probability proportional to their squared distance
* to the current cluster set. It results in a provable approximation to an optimal clustering.
*
* The original paper can be found at http://theory.stanford.edu/~sergei/papers/vldb12-kmpar.pdf.
*/
- private def initKMeansParallel(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
+ private def initKMeansParallel(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
// Initialize empty centers and point costs.
- val centers = Array.tabulate(runs)(r => ArrayBuffer.empty[VectorWithNorm])
- var costs = data.map(_ => Array.fill(runs)(Double.PositiveInfinity))
+ var costs = data.map(_ => Double.PositiveInfinity)
- // Initialize each run's first center to a random point.
+ // Initialize the first center to a random point.
val seed = new XORShiftRandom(this.seed).nextInt()
- val sample = data.takeSample(true, runs, seed).toSeq
+ val sample = data.takeSample(false, 1, seed)
// Could be empty if data is empty; fail with a better message early:
- require(sample.size >= runs, s"Required $runs samples but got ${sample.size} from $data")
- val newCenters = Array.tabulate(runs)(r => ArrayBuffer(sample(r).toDense))
-
- /** Merges new centers to centers. */
- def mergeNewCenters(): Unit = {
- var r = 0
- while (r < runs) {
- centers(r) ++= newCenters(r)
- newCenters(r).clear()
- r += 1
- }
- }
+ require(sample.nonEmpty, s"No samples available from $data")
- // On each step, sample 2 * k points on average for each run with probability proportional
- // to their squared distance from that run's centers. Note that only distances between points
+ val centers = ArrayBuffer[VectorWithNorm]()
+ var newCenters = Seq(sample.head.toDense)
--- End diff --
Why is it faster if it's an Array instead of Seq? or am I getting the comments cross-wired?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMean...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r81848814
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,107 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var active = true
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
-
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && active) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ active = false
+ totalContribs.foreach { case (j, (sum, count)) =>
+ scal(1.0 / count, sum)
+ val newCenter = new VectorWithNorm(sum)
+ if (!active && KMeans.fastSquaredDistance(newCenter, centers(j)) > epsilon * epsilon) {
+ active = true
}
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
- }
- costs(run) = costAccums(i).value
+ centers(j) = newCenter
}
- activeRuns = activeRuns.filter(active(_))
+ cost = costAccum.value
iteration += 1
}
val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
- logInfo(s"Iterations took " + "%.3f".format(iterationTimeInSeconds) + " seconds.")
+ logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")
if (iteration == maxIterations) {
logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
} else {
logInfo(s"KMeans converged in $iteration iterations.")
}
- val (minCost, bestRun) = costs.zipWithIndex.min
-
- logInfo(s"The cost for the best run is $minCost.")
+ logInfo(s"The cost is $cost.")
- new KMeansModel(centers(bestRun).map(_.vector))
+ new KMeansModel(centers.map(_.vector))
}
/**
- * Initialize `runs` sets of cluster centers at random.
+ * Initialize set of cluster centers at random.
*/
- private def initRandom(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
- // Sample all the cluster centers in one pass to avoid repeated scans
- val sample = data.takeSample(true, runs * k, new XORShiftRandom(this.seed).nextInt()).toSeq
- Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
- new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm)
- }.toArray)
+ private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
+ val sample = data.takeSample(false, k, new XORShiftRandom(this.seed).nextInt())
+ sample.map(v => new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm))
}
/**
- * Initialize `runs` sets of cluster centers using the k-means|| algorithm by Bahmani et al.
+ * Initialize set of cluster centers using the k-means|| algorithm by Bahmani et al.
* (Bahmani et al., Scalable K-Means++, VLDB 2012). This is a variant of k-means++ that tries
* to find with dissimilar cluster centers by starting with a random center and then doing
--- End diff --
"to find with dissimilar" -> "to find dissimilar" (while we're here)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82691339
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -531,6 +471,7 @@ object KMeans {
* "k-means||". (default: "k-means||")
*/
@Since("0.8.0")
+ @deprecated("Use train method without 'runs'", "2.1.0")
def train(
--- End diff --
This signature does not have a direct alternative without `runs`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15342
**[Test build #66578 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66578/consoleFull)** for PR 15342 at commit [`68e3d90`](https://github.com/apache/spark/commit/68e3d90a790060f496cdbfeea8171e5f531633e8).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15342
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82633045
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
+ totalContribs.foreach { case (j, (sum, count)) =>
+ scal(1.0 / count, sum)
+ val newCenter = new VectorWithNorm(sum)
+ if (converged && KMeans.fastSquaredDistance(newCenter, centers(j)) > epsilon * epsilon) {
+ converged = false
}
- costs(run) = costAccums(i).value
+ centers(j) = newCenter
}
- activeRuns = activeRuns.filter(active(_))
+ cost = costAccum.value
iteration += 1
}
val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
- logInfo(s"Iterations took " + "%.3f".format(iterationTimeInSeconds) + " seconds.")
+ logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")
if (iteration == maxIterations) {
logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
} else {
logInfo(s"KMeans converged in $iteration iterations.")
}
- val (minCost, bestRun) = costs.zipWithIndex.min
+ logInfo(s"The cost is $cost.")
- logInfo(s"The cost for the best run is $minCost.")
-
- new KMeansModel(centers(bestRun).map(_.vector))
+ new KMeansModel(centers.map(_.vector))
}
/**
- * Initialize `runs` sets of cluster centers at random.
+ * Initialize a set of cluster centers at random.
*/
- private def initRandom(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
- // Sample all the cluster centers in one pass to avoid repeated scans
- val sample = data.takeSample(true, runs * k, new XORShiftRandom(this.seed).nextInt()).toSeq
- Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
- new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm)
- }.toArray)
+ private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
+ data.takeSample(true, k, new XORShiftRandom(this.seed).nextInt()).map(_.toDense)
--- End diff --
I guess not, but the centers become immediately dense in the first iteration of `runAlgorithm`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMean...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r81848024
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,107 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var active = true
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
-
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && active) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ active = false
+ totalContribs.foreach { case (j, (sum, count)) =>
+ scal(1.0 / count, sum)
+ val newCenter = new VectorWithNorm(sum)
+ if (!active && KMeans.fastSquaredDistance(newCenter, centers(j)) > epsilon * epsilon) {
+ active = true
}
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
- }
- costs(run) = costAccums(i).value
+ centers(j) = newCenter
}
- activeRuns = activeRuns.filter(active(_))
+ cost = costAccum.value
iteration += 1
}
val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
- logInfo(s"Iterations took " + "%.3f".format(iterationTimeInSeconds) + " seconds.")
+ logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")
if (iteration == maxIterations) {
logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
} else {
logInfo(s"KMeans converged in $iteration iterations.")
}
- val (minCost, bestRun) = costs.zipWithIndex.min
-
- logInfo(s"The cost for the best run is $minCost.")
+ logInfo(s"The cost is $cost.")
- new KMeansModel(centers(bestRun).map(_.vector))
+ new KMeansModel(centers.map(_.vector))
}
/**
- * Initialize `runs` sets of cluster centers at random.
+ * Initialize set of cluster centers at random.
*/
- private def initRandom(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
- // Sample all the cluster centers in one pass to avoid repeated scans
- val sample = data.takeSample(true, runs * k, new XORShiftRandom(this.seed).nextInt()).toSeq
- Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
- new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm)
- }.toArray)
+ private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
+ val sample = data.takeSample(false, k, new XORShiftRandom(this.seed).nextInt())
+ sample.map(v => new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm))
--- End diff --
`sample.map(_.toDense)`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15342
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/66310/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15342
**[Test build #66729 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66729/consoleFull)** for PR 15342 at commit [`ba52582`](https://github.com/apache/spark/commit/ba52582a1313be9d9febe215fe6f21b2d9be239f).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15342
**[Test build #66578 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66578/consoleFull)** for PR 15342 at commit [`68e3d90`](https://github.com/apache/spark/commit/68e3d90a790060f496cdbfeea8171e5f531633e8).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15342
**[Test build #66310 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66310/consoleFull)** for PR 15342 at commit [`cd14b65`](https://github.com/apache/spark/commit/cd14b65ac082d728d3a99aeede54c9a7e1f22614).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by yanboliang <gi...@git.apache.org>.
Github user yanboliang commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82625915
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -409,74 +360,39 @@ class KMeans private (
bcNewCentersList += bcNewCenters
val preCosts = costs
costs = data.zip(preCosts).map { case (point, cost) =>
- Array.tabulate(runs) { r =>
- math.min(KMeans.pointCost(bcNewCenters.value(r), point), cost(r))
- }
- }.persist(StorageLevel.MEMORY_AND_DISK)
- val sumCosts = costs
- .aggregate(new Array[Double](runs))(
- seqOp = (s, v) => {
- // s += v
- var r = 0
- while (r < runs) {
- s(r) += v(r)
- r += 1
- }
- s
- },
- combOp = (s0, s1) => {
- // s0 += s1
- var r = 0
- while (r < runs) {
- s0(r) += s1(r)
- r += 1
- }
- s0
- }
- )
+ math.min(KMeans.pointCost(bcNewCenters.value, point), cost)
+ }.persist(StorageLevel.MEMORY_AND_DISK)
+ val sumCosts = costs.sum()
bcNewCenters.unpersist(blocking = false)
preCosts.unpersist(blocking = false)
- val chosen = data.zip(costs).mapPartitionsWithIndex { (index, pointsWithCosts) =>
+ val chosen = data.zip(costs).mapPartitionsWithIndex { (index, pointCosts) =>
val rand = new XORShiftRandom(seed ^ (step << 16) ^ index)
- pointsWithCosts.flatMap { case (p, c) =>
- val rs = (0 until runs).filter { r =>
- rand.nextDouble() < 2.0 * c(r) * k / sumCosts(r)
- }
- if (rs.nonEmpty) Some((p, rs)) else None
- }
+ pointCosts.filter { case (_, c) => rand.nextDouble() < 2.0 * c * k / sumCosts }.map(_._1)
}.collect()
- mergeNewCenters()
- chosen.foreach { case (p, rs) =>
- rs.foreach(newCenters(_) += p.toDense)
- }
+ newCenters = chosen.map(_.toDense)
--- End diff --
Ditto.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/15342
@sethah OK I will add a new overload of `train` and deprecate the others.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/15342
Merged to master. I'm going to reopen a PR for just the duplicate centroids issue to re-table that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMean...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r81849054
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,107 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var active = true
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
-
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && active) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ active = false
+ totalContribs.foreach { case (j, (sum, count)) =>
+ scal(1.0 / count, sum)
+ val newCenter = new VectorWithNorm(sum)
+ if (!active && KMeans.fastSquaredDistance(newCenter, centers(j)) > epsilon * epsilon) {
+ active = true
}
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
- }
- costs(run) = costAccums(i).value
+ centers(j) = newCenter
}
- activeRuns = activeRuns.filter(active(_))
+ cost = costAccum.value
iteration += 1
}
val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
- logInfo(s"Iterations took " + "%.3f".format(iterationTimeInSeconds) + " seconds.")
+ logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")
if (iteration == maxIterations) {
logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
} else {
logInfo(s"KMeans converged in $iteration iterations.")
}
- val (minCost, bestRun) = costs.zipWithIndex.min
-
- logInfo(s"The cost for the best run is $minCost.")
+ logInfo(s"The cost is $cost.")
- new KMeansModel(centers(bestRun).map(_.vector))
+ new KMeansModel(centers.map(_.vector))
}
/**
- * Initialize `runs` sets of cluster centers at random.
+ * Initialize set of cluster centers at random.
*/
- private def initRandom(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
- // Sample all the cluster centers in one pass to avoid repeated scans
- val sample = data.takeSample(true, runs * k, new XORShiftRandom(this.seed).nextInt()).toSeq
- Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
- new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm)
- }.toArray)
+ private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
+ val sample = data.takeSample(false, k, new XORShiftRandom(this.seed).nextInt())
+ sample.map(v => new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm))
}
/**
- * Initialize `runs` sets of cluster centers using the k-means|| algorithm by Bahmani et al.
+ * Initialize set of cluster centers using the k-means|| algorithm by Bahmani et al.
* (Bahmani et al., Scalable K-Means++, VLDB 2012). This is a variant of k-means++ that tries
* to find with dissimilar cluster centers by starting with a random center and then doing
* passes where more centers are chosen with probability proportional to their squared distance
* to the current cluster set. It results in a provable approximation to an optimal clustering.
*
* The original paper can be found at http://theory.stanford.edu/~sergei/papers/vldb12-kmpar.pdf.
*/
- private def initKMeansParallel(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
+ private def initKMeansParallel(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
// Initialize empty centers and point costs.
- val centers = Array.tabulate(runs)(r => ArrayBuffer.empty[VectorWithNorm])
- var costs = data.map(_ => Array.fill(runs)(Double.PositiveInfinity))
+ var costs = data.map(_ => Double.PositiveInfinity)
// Initialize each run's first center to a random point.
--- End diff --
"Initialize the first center to a random point."
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/15342
I'm wondering, what's the use case for allowing duplicate centroids? it doesn't have a reasonable meaning and does slow down execution. I don't feel _so_ strongly about it and would like to get the change to remove "runs" in regardless, so could back that out, but I'd be a little more convinced if it were more than just matching scikit
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15342
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by yanboliang <gi...@git.apache.org>.
Github user yanboliang commented on the issue:
https://github.com/apache/spark/pull/15342
Only the last two minor items, otherwise, this looks ready to me. Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15342
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82649829
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
+ totalContribs.foreach { case (j, (sum, count)) =>
+ scal(1.0 / count, sum)
+ val newCenter = new VectorWithNorm(sum)
+ if (converged && KMeans.fastSquaredDistance(newCenter, centers(j)) > epsilon * epsilon) {
+ converged = false
}
- costs(run) = costAccums(i).value
+ centers(j) = newCenter
}
- activeRuns = activeRuns.filter(active(_))
+ cost = costAccum.value
iteration += 1
}
val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
- logInfo(s"Iterations took " + "%.3f".format(iterationTimeInSeconds) + " seconds.")
+ logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")
if (iteration == maxIterations) {
logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
} else {
logInfo(s"KMeans converged in $iteration iterations.")
}
- val (minCost, bestRun) = costs.zipWithIndex.min
+ logInfo(s"The cost is $cost.")
- logInfo(s"The cost for the best run is $minCost.")
-
- new KMeansModel(centers(bestRun).map(_.vector))
+ new KMeansModel(centers.map(_.vector))
}
/**
- * Initialize `runs` sets of cluster centers at random.
+ * Initialize a set of cluster centers at random.
*/
- private def initRandom(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
- // Sample all the cluster centers in one pass to avoid repeated scans
- val sample = data.takeSample(true, runs * k, new XORShiftRandom(this.seed).nextInt()).toSeq
- Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
- new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm)
- }.toArray)
+ private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
+ data.takeSample(true, k, new XORShiftRandom(this.seed).nextInt()).map(_.toDense)
--- End diff --
At least, it's what the existing code did.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15342
**[Test build #66381 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66381/consoleFull)** for PR 15342 at commit [`ebbb852`](https://github.com/apache/spark/commit/ebbb852ad69f3a0a9c9facbd4a55ecce0eb86df0).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/15342
This is what SPARK-3261 is about. It's a corner case to be sure. To me it seems like having duplicate centroids is worse because the model loses some of its meaning. Points may arbitrarily assign to one or the other of two identical centroids. Of the 3 possible behaviors, looks like we have all 3 on the table:
1) error
2) return < k centroids
3) return k centroids
I suppose I prefer the new behavior but I can't say I feel that strongly. I guess matching scikit has some value.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82652315
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
--- End diff --
Unless I'm overlooking some obviously nicer expression, I think the loop is going to work the same either way: you have to assume you terminate unless a distance proves otherwise, per iteration.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15342
**[Test build #66673 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66673/consoleFull)** for PR 15342 at commit [`5cb9e5f`](https://github.com/apache/spark/commit/5cb9e5f5438c94e3f44545ab65f8ccd7a85f55af).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82687576
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
--- End diff --
Yep, you're correct. Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15342
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/66578/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82646051
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -558,6 +475,7 @@ object KMeans {
* Trains a k-means model using specified parameters and the default values for unspecified.
*/
@Since("0.8.0")
+ @deprecated("Use train method without 'runs'", "2.1.0")
--- End diff --
I think we should add them for completeness, and deprecate all overloads using `runs`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82653577
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
--- End diff --
The logic is the same, yes, but it seems really strange to set something to false, then each iteration set it to true and then set it back false if some condition. Why not leave it false and change to true if convergence criteria is met? This is basically a trivial detail, so only change it if you want. I'm fine either way.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15342
**[Test build #66684 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66684/consoleFull)** for PR 15342 at commit [`84fb22f`](https://github.com/apache/spark/commit/84fb22f9817a2d7660eaae44442a6e88fd8471c1).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by yanboliang <gi...@git.apache.org>.
Github user yanboliang commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82625743
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
+ totalContribs.foreach { case (j, (sum, count)) =>
+ scal(1.0 / count, sum)
+ val newCenter = new VectorWithNorm(sum)
+ if (converged && KMeans.fastSquaredDistance(newCenter, centers(j)) > epsilon * epsilon) {
+ converged = false
}
- costs(run) = costAccums(i).value
+ centers(j) = newCenter
}
- activeRuns = activeRuns.filter(active(_))
+ cost = costAccum.value
iteration += 1
}
val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
- logInfo(s"Iterations took " + "%.3f".format(iterationTimeInSeconds) + " seconds.")
+ logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")
if (iteration == maxIterations) {
logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
} else {
logInfo(s"KMeans converged in $iteration iterations.")
}
- val (minCost, bestRun) = costs.zipWithIndex.min
+ logInfo(s"The cost is $cost.")
- logInfo(s"The cost for the best run is $minCost.")
-
- new KMeansModel(centers(bestRun).map(_.vector))
+ new KMeansModel(centers.map(_.vector))
}
/**
- * Initialize `runs` sets of cluster centers at random.
+ * Initialize a set of cluster centers at random.
*/
- private def initRandom(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
- // Sample all the cluster centers in one pass to avoid repeated scans
- val sample = data.takeSample(true, runs * k, new XORShiftRandom(this.seed).nextInt()).toSeq
- Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
- new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm)
- }.toArray)
+ private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
+ data.takeSample(true, k, new XORShiftRandom(this.seed).nextInt()).map(_.toDense)
}
/**
- * Initialize `runs` sets of cluster centers using the k-means|| algorithm by Bahmani et al.
+ * Initialize a set of cluster centers using the k-means|| algorithm by Bahmani et al.
* (Bahmani et al., Scalable K-Means++, VLDB 2012). This is a variant of k-means++ that tries
- * to find with dissimilar cluster centers by starting with a random center and then doing
+ * to find dissimilar cluster centers by starting with a random center and then doing
* passes where more centers are chosen with probability proportional to their squared distance
* to the current cluster set. It results in a provable approximation to an optimal clustering.
*
* The original paper can be found at http://theory.stanford.edu/~sergei/papers/vldb12-kmpar.pdf.
*/
- private def initKMeansParallel(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
+ private def initKMeansParallel(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
// Initialize empty centers and point costs.
- val centers = Array.tabulate(runs)(r => ArrayBuffer.empty[VectorWithNorm])
- var costs = data.map(_ => Array.fill(runs)(Double.PositiveInfinity))
+ var costs = data.map(_ => Double.PositiveInfinity)
- // Initialize each run's first center to a random point.
+ // Initialize the first center to a random point.
val seed = new XORShiftRandom(this.seed).nextInt()
- val sample = data.takeSample(true, runs, seed).toSeq
+ val sample = data.takeSample(false, 1, seed)
// Could be empty if data is empty; fail with a better message early:
- require(sample.size >= runs, s"Required $runs samples but got ${sample.size} from $data")
- val newCenters = Array.tabulate(runs)(r => ArrayBuffer(sample(r).toDense))
-
- /** Merges new centers to centers. */
- def mergeNewCenters(): Unit = {
- var r = 0
- while (r < runs) {
- centers(r) ++= newCenters(r)
- newCenters(r).clear()
- r += 1
- }
- }
+ require(sample.nonEmpty, s"No samples available from $data")
- // On each step, sample 2 * k points on average for each run with probability proportional
- // to their squared distance from that run's centers. Note that only distances between points
+ val centers = ArrayBuffer[VectorWithNorm]()
+ var newCenters = Seq(sample.head.toDense)
--- End diff --
Maybe irrelevant with this PR, why we need to cast it to dense one?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on the issue:
https://github.com/apache/spark/pull/15342
LGTM
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/15342
@sethah are you OK with this part? We can still talk about the k centroids bit, either here or on the JIRA.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by srowen <gi...@git.apache.org>.
Github user srowen closed the pull request at:
https://github.com/apache/spark/pull/15342
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15342
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/15342
Good question. I think he's saying that it returned 1K centers _after_ this change. It's a good point that this would also speed things up considerably, because computing the distance to duplicate centroids is all superfluous work.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15342
**[Test build #66673 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66673/consoleFull)** for PR 15342 at commit [`5cb9e5f`](https://github.com/apache/spark/commit/5cb9e5f5438c94e3f44545ab65f8ccd7a85f55af).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by yanboliang <gi...@git.apache.org>.
Github user yanboliang commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82619472
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -43,21 +43,21 @@ import org.apache.spark.util.random.XORShiftRandom
class KMeans private (
private var k: Int,
private var maxIterations: Int,
- private var runs: Int,
private var initializationMode: String,
private var initializationSteps: Int,
private var epsilon: Double,
private var seed: Long) extends Serializable with Logging {
/**
- * Constructs a KMeans instance with default parameters: {k: 2, maxIterations: 20, runs: 1,
+ * Constructs a KMeans instance with default parameters: {k: 2, maxIterations: 20,
* initializationMode: "k-means||", initializationSteps: 2, epsilon: 1e-4, seed: random}.
*/
@Since("0.8.0")
- def this() = this(2, 20, 1, KMeans.K_MEANS_PARALLEL, 2, 1e-4, Utils.random.nextLong())
+ def this() = this(2, 20, KMeans.K_MEANS_PARALLEL, 2, 1e-4, Utils.random.nextLong())
/**
- * Number of clusters to create (k).
+ * Number of clusters to create (k). Note that if the input has fewer than k elements,
+ * then it's possible that fewer than k clusters are created.
--- End diff --
If we back out change to avoid duplicate centroids, does this annotation be invalid?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by yanboliang <gi...@git.apache.org>.
Github user yanboliang commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82626968
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
--- End diff --
I think ```changed``` would be more intuitive.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15342
**[Test build #66310 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66310/consoleFull)** for PR 15342 at commit [`cd14b65`](https://github.com/apache/spark/commit/cd14b65ac082d728d3a99aeede54c9a7e1f22614).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15342
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82654945
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
+ totalContribs.foreach { case (j, (sum, count)) =>
+ scal(1.0 / count, sum)
+ val newCenter = new VectorWithNorm(sum)
+ if (converged && KMeans.fastSquaredDistance(newCenter, centers(j)) > epsilon * epsilon) {
+ converged = false
}
- costs(run) = costAccums(i).value
+ centers(j) = newCenter
}
- activeRuns = activeRuns.filter(active(_))
+ cost = costAccum.value
iteration += 1
}
val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
- logInfo(s"Iterations took " + "%.3f".format(iterationTimeInSeconds) + " seconds.")
+ logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")
if (iteration == maxIterations) {
logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
} else {
logInfo(s"KMeans converged in $iteration iterations.")
}
- val (minCost, bestRun) = costs.zipWithIndex.min
+ logInfo(s"The cost is $cost.")
- logInfo(s"The cost for the best run is $minCost.")
-
- new KMeansModel(centers(bestRun).map(_.vector))
+ new KMeansModel(centers.map(_.vector))
}
/**
- * Initialize `runs` sets of cluster centers at random.
+ * Initialize a set of cluster centers at random.
*/
- private def initRandom(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
- // Sample all the cluster centers in one pass to avoid repeated scans
- val sample = data.takeSample(true, runs * k, new XORShiftRandom(this.seed).nextInt()).toSeq
- Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
- new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm)
- }.toArray)
+ private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
+ data.takeSample(true, k, new XORShiftRandom(this.seed).nextInt()).map(_.toDense)
}
/**
- * Initialize `runs` sets of cluster centers using the k-means|| algorithm by Bahmani et al.
+ * Initialize a set of cluster centers using the k-means|| algorithm by Bahmani et al.
* (Bahmani et al., Scalable K-Means++, VLDB 2012). This is a variant of k-means++ that tries
- * to find with dissimilar cluster centers by starting with a random center and then doing
+ * to find dissimilar cluster centers by starting with a random center and then doing
* passes where more centers are chosen with probability proportional to their squared distance
* to the current cluster set. It results in a provable approximation to an optimal clustering.
*
* The original paper can be found at http://theory.stanford.edu/~sergei/papers/vldb12-kmpar.pdf.
*/
- private def initKMeansParallel(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
+ private def initKMeansParallel(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
// Initialize empty centers and point costs.
- val centers = Array.tabulate(runs)(r => ArrayBuffer.empty[VectorWithNorm])
- var costs = data.map(_ => Array.fill(runs)(Double.PositiveInfinity))
+ var costs = data.map(_ => Double.PositiveInfinity)
- // Initialize each run's first center to a random point.
+ // Initialize the first center to a random point.
val seed = new XORShiftRandom(this.seed).nextInt()
- val sample = data.takeSample(true, runs, seed).toSeq
+ val sample = data.takeSample(false, 1, seed)
// Could be empty if data is empty; fail with a better message early:
- require(sample.size >= runs, s"Required $runs samples but got ${sample.size} from $data")
- val newCenters = Array.tabulate(runs)(r => ArrayBuffer(sample(r).toDense))
-
- /** Merges new centers to centers. */
- def mergeNewCenters(): Unit = {
- var r = 0
- while (r < runs) {
- centers(r) ++= newCenters(r)
- newCenters(r).clear()
- r += 1
- }
- }
+ require(sample.nonEmpty, s"No samples available from $data")
- // On each step, sample 2 * k points on average for each run with probability proportional
- // to their squared distance from that run's centers. Note that only distances between points
+ val centers = ArrayBuffer[VectorWithNorm]()
+ var newCenters = Seq(sample.head.toDense)
--- End diff --
The difference is probably negligible. I just thought we could use `Array` if there is no specific preference for using a `List`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15342
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/66729/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on the issue:
https://github.com/apache/spark/pull/15342
Looking good. My main concern is that now you can have the following:
````scala
scala> model.getK
res2: Int = 3
scala> model.clusterCenters.length
res3: Int = 1
````
We could set the model `k` to match the cluster centers length before creating the model, during training. We could leave it, but then what does `k` mean, if not the number of centers?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by yanboliang <gi...@git.apache.org>.
Github user yanboliang commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82627280
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
+ totalContribs.foreach { case (j, (sum, count)) =>
--- End diff --
Compared with the original code, ```foreach``` may slower than ```while``` loop if you have a large ```k```.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/15342
Yeah, but now we have yet 2 more overloads. I had intended to point people to 1 new overload, but I guess it's weird to make people specify the seed arg. And optional args, the normal solution, breaks binary compatibility IIRC
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/15342
CC @yanboliang
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15342
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/66673/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15342
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/66381/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMean...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r81848246
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,107 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var active = true
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
-
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && active) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ active = false
+ totalContribs.foreach { case (j, (sum, count)) =>
+ scal(1.0 / count, sum)
+ val newCenter = new VectorWithNorm(sum)
+ if (!active && KMeans.fastSquaredDistance(newCenter, centers(j)) > epsilon * epsilon) {
+ active = true
}
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
- }
- costs(run) = costAccums(i).value
+ centers(j) = newCenter
}
- activeRuns = activeRuns.filter(active(_))
+ cost = costAccum.value
iteration += 1
}
val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
- logInfo(s"Iterations took " + "%.3f".format(iterationTimeInSeconds) + " seconds.")
+ logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")
if (iteration == maxIterations) {
logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
} else {
logInfo(s"KMeans converged in $iteration iterations.")
}
- val (minCost, bestRun) = costs.zipWithIndex.min
-
- logInfo(s"The cost for the best run is $minCost.")
+ logInfo(s"The cost is $cost.")
- new KMeansModel(centers(bestRun).map(_.vector))
+ new KMeansModel(centers.map(_.vector))
}
/**
- * Initialize `runs` sets of cluster centers at random.
+ * Initialize set of cluster centers at random.
*/
- private def initRandom(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
- // Sample all the cluster centers in one pass to avoid repeated scans
- val sample = data.takeSample(true, runs * k, new XORShiftRandom(this.seed).nextInt()).toSeq
- Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
- new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm)
- }.toArray)
+ private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
+ val sample = data.takeSample(false, k, new XORShiftRandom(this.seed).nextInt())
+ sample.map(v => new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm))
}
/**
- * Initialize `runs` sets of cluster centers using the k-means|| algorithm by Bahmani et al.
+ * Initialize set of cluster centers using the k-means|| algorithm by Bahmani et al.
--- End diff --
nit: "Initialize a set"
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on the issue:
https://github.com/apache/spark/pull/15342
@srowen I will take a look shortly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82648980
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
+ totalContribs.foreach { case (j, (sum, count)) =>
--- End diff --
Why is that? I'm aware that Scala `for` comprehensions can desugar into something surprisingly expensive, but this seems clearer and about the same as a `while`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82648993
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -43,21 +43,21 @@ import org.apache.spark.util.random.XORShiftRandom
class KMeans private (
private var k: Int,
private var maxIterations: Int,
- private var runs: Int,
private var initializationMode: String,
private var initializationSteps: Int,
private var epsilon: Double,
private var seed: Long) extends Serializable with Logging {
/**
- * Constructs a KMeans instance with default parameters: {k: 2, maxIterations: 20, runs: 1,
+ * Constructs a KMeans instance with default parameters: {k: 2, maxIterations: 20,
* initializationMode: "k-means||", initializationSteps: 2, epsilon: 1e-4, seed: random}.
*/
@Since("0.8.0")
- def this() = this(2, 20, 1, KMeans.K_MEANS_PARALLEL, 2, 1e-4, Utils.random.nextLong())
+ def this() = this(2, 20, KMeans.K_MEANS_PARALLEL, 2, 1e-4, Utils.random.nextLong())
/**
- * Number of clusters to create (k).
+ * Number of clusters to create (k). Note that if the input has fewer than k elements,
+ * then it's possible that fewer than k clusters are created.
--- End diff --
Ooops, right
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15342
**[Test build #66684 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66684/consoleFull)** for PR 15342 at commit [`84fb22f`](https://github.com/apache/spark/commit/84fb22f9817a2d7660eaae44442a6e88fd8471c1).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82634889
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
+ totalContribs.foreach { case (j, (sum, count)) =>
+ scal(1.0 / count, sum)
+ val newCenter = new VectorWithNorm(sum)
+ if (converged && KMeans.fastSquaredDistance(newCenter, centers(j)) > epsilon * epsilon) {
+ converged = false
}
- costs(run) = costAccums(i).value
+ centers(j) = newCenter
}
- activeRuns = activeRuns.filter(active(_))
+ cost = costAccum.value
iteration += 1
}
val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
- logInfo(s"Iterations took " + "%.3f".format(iterationTimeInSeconds) + " seconds.")
+ logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")
if (iteration == maxIterations) {
logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
} else {
logInfo(s"KMeans converged in $iteration iterations.")
}
- val (minCost, bestRun) = costs.zipWithIndex.min
+ logInfo(s"The cost is $cost.")
- logInfo(s"The cost for the best run is $minCost.")
-
- new KMeansModel(centers(bestRun).map(_.vector))
+ new KMeansModel(centers.map(_.vector))
}
/**
- * Initialize `runs` sets of cluster centers at random.
+ * Initialize a set of cluster centers at random.
*/
- private def initRandom(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
- // Sample all the cluster centers in one pass to avoid repeated scans
- val sample = data.takeSample(true, runs * k, new XORShiftRandom(this.seed).nextInt()).toSeq
- Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
- new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm)
- }.toArray)
+ private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
+ data.takeSample(true, k, new XORShiftRandom(this.seed).nextInt()).map(_.toDense)
}
/**
- * Initialize `runs` sets of cluster centers using the k-means|| algorithm by Bahmani et al.
+ * Initialize a set of cluster centers using the k-means|| algorithm by Bahmani et al.
* (Bahmani et al., Scalable K-Means++, VLDB 2012). This is a variant of k-means++ that tries
- * to find with dissimilar cluster centers by starting with a random center and then doing
+ * to find dissimilar cluster centers by starting with a random center and then doing
* passes where more centers are chosen with probability proportional to their squared distance
* to the current cluster set. It results in a provable approximation to an optimal clustering.
*
* The original paper can be found at http://theory.stanford.edu/~sergei/papers/vldb12-kmpar.pdf.
*/
- private def initKMeansParallel(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
+ private def initKMeansParallel(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
// Initialize empty centers and point costs.
- val centers = Array.tabulate(runs)(r => ArrayBuffer.empty[VectorWithNorm])
- var costs = data.map(_ => Array.fill(runs)(Double.PositiveInfinity))
+ var costs = data.map(_ => Double.PositiveInfinity)
- // Initialize each run's first center to a random point.
+ // Initialize the first center to a random point.
val seed = new XORShiftRandom(this.seed).nextInt()
- val sample = data.takeSample(true, runs, seed).toSeq
+ val sample = data.takeSample(false, 1, seed)
// Could be empty if data is empty; fail with a better message early:
- require(sample.size >= runs, s"Required $runs samples but got ${sample.size} from $data")
- val newCenters = Array.tabulate(runs)(r => ArrayBuffer(sample(r).toDense))
-
- /** Merges new centers to centers. */
- def mergeNewCenters(): Unit = {
- var r = 0
- while (r < runs) {
- centers(r) ++= newCenters(r)
- newCenters(r).clear()
- r += 1
- }
- }
+ require(sample.nonEmpty, s"No samples available from $data")
- // On each step, sample 2 * k points on average for each run with probability proportional
- // to their squared distance from that run's centers. Note that only distances between points
+ val centers = ArrayBuffer[VectorWithNorm]()
+ var newCenters = Seq(sample.head.toDense)
--- End diff --
I'm not sure this is performance critical, but `centers ++= newCenters` will be faster if `newCenters` is an Array instead of List.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15342
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15342
**[Test build #66729 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66729/consoleFull)** for PR 15342 at commit [`ba52582`](https://github.com/apache/spark/commit/ba52582a1313be9d9febe215fe6f21b2d9be239f).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by yanboliang <gi...@git.apache.org>.
Github user yanboliang commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82734529
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
+ totalContribs.foreach { case (j, (sum, count)) =>
+ scal(1.0 / count, sum)
+ val newCenter = new VectorWithNorm(sum)
+ if (converged && KMeans.fastSquaredDistance(newCenter, centers(j)) > epsilon * epsilon) {
+ converged = false
}
- costs(run) = costAccums(i).value
+ centers(j) = newCenter
}
- activeRuns = activeRuns.filter(active(_))
+ cost = costAccum.value
iteration += 1
}
val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
- logInfo(s"Iterations took " + "%.3f".format(iterationTimeInSeconds) + " seconds.")
+ logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")
if (iteration == maxIterations) {
logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
} else {
logInfo(s"KMeans converged in $iteration iterations.")
}
- val (minCost, bestRun) = costs.zipWithIndex.min
+ logInfo(s"The cost is $cost.")
- logInfo(s"The cost for the best run is $minCost.")
-
- new KMeansModel(centers(bestRun).map(_.vector))
+ new KMeansModel(centers.map(_.vector))
}
/**
- * Initialize `runs` sets of cluster centers at random.
+ * Initialize a set of cluster centers at random.
*/
- private def initRandom(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
- // Sample all the cluster centers in one pass to avoid repeated scans
- val sample = data.takeSample(true, runs * k, new XORShiftRandom(this.seed).nextInt()).toSeq
- Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
- new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm)
- }.toArray)
+ private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
+ data.takeSample(true, k, new XORShiftRandom(this.seed).nextInt()).map(_.toDense)
--- End diff --
Maybe we can optimize this issue at #14937, since it will have different treatment for dense and sparse vector.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82650874
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
+ totalContribs.foreach { case (j, (sum, count)) =>
--- End diff --
In general `while` is faster than `foreach` (creating and calling an anonymous function), but I'd be surprised if it affected performance here because we are only running this once per iteration and the bulk of the cost will be distributed computation.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/15342
That's right. `k` seems like the requested number of centroids, which may not match the actual number in corner cases. What about just documenting that more?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by yanboliang <gi...@git.apache.org>.
Github user yanboliang commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82736118
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -531,6 +471,7 @@ object KMeans {
* "k-means||". (default: "k-means||")
*/
@Since("0.8.0")
+ @deprecated("Use train method without 'runs'", "2.1.0")
def train(
--- End diff --
+1 @sethah
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on the issue:
https://github.com/apache/spark/pull/15342
@srowen That is not the impression that I got from "I just ran clustering on 1.3M points, asking for 10,000 clusters. This clustering run resulted in 1019 unique cluster centers."
@derrickburns Can you clarify a bit here? Also, could you tell us the nature of the data that was used for your clustering?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82625386
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
--- End diff --
Why don't we just leave converged false, and only change it to true inside the foreach?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMean...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r81864909
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,107 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var active = true
--- End diff --
I find `converged` to be more intuitive, but not a strong preference.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementation / r...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15342
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/66684/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82690396
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -499,13 +414,38 @@ object KMeans {
* @param data Training points as an `RDD` of `Vector` types.
* @param k Number of clusters to create.
* @param maxIterations Maximum number of iterations allowed.
+ * @param initializationMode The initialization algorithm. This can either be "random" or
+ * "k-means||". (default: "k-means||")
+ * @param seed Random seed for cluster initialization. Default is to generate seed based
+ * on system time.
+ */
+ @Since("2.1.0")
+ def train(data: RDD[Vector],
--- End diff --
minor: style should match other train signatures
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82682188
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
--- End diff --
I don't think it can be done the way you're suggesting; it's not just preference. You could just set it with a nice simple call `.forall` as you're suggesting, usually, but here we also need the side effect of visiting each element. To do both I think we have to 'unroll' the equivalent logic and it amounts to this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMean...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r81913935
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -558,6 +475,7 @@ object KMeans {
* Trains a k-means model using specified parameters and the default values for unspecified.
*/
@Since("0.8.0")
+ @deprecated("Use train method without 'runs'", "2.1.0")
--- End diff --
Yes, though there's no alternative to those with the same arguments. We could add another overload and deprecate the others. I'm OK with that too, just felt a little gross to add yet more.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMean...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r81862372
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -409,74 +361,39 @@ class KMeans private (
bcNewCentersList += bcNewCenters
val preCosts = costs
costs = data.zip(preCosts).map { case (point, cost) =>
- Array.tabulate(runs) { r =>
- math.min(KMeans.pointCost(bcNewCenters.value(r), point), cost(r))
- }
- }.persist(StorageLevel.MEMORY_AND_DISK)
- val sumCosts = costs
- .aggregate(new Array[Double](runs))(
- seqOp = (s, v) => {
- // s += v
- var r = 0
- while (r < runs) {
- s(r) += v(r)
- r += 1
- }
- s
- },
- combOp = (s0, s1) => {
- // s0 += s1
- var r = 0
- while (r < runs) {
- s0(r) += s1(r)
- r += 1
- }
- s0
- }
- )
+ math.min(KMeans.pointCost(bcNewCenters.value, point), cost)
+ }.persist(StorageLevel.MEMORY_AND_DISK)
+ val sumCosts = costs.sum()
bcNewCenters.unpersist(blocking = false)
preCosts.unpersist(blocking = false)
- val chosen = data.zip(costs).mapPartitionsWithIndex { (index, pointsWithCosts) =>
+ val chosen = data.zip(costs).mapPartitionsWithIndex { (index, pointCosts) =>
val rand = new XORShiftRandom(seed ^ (step << 16) ^ index)
- pointsWithCosts.flatMap { case (p, c) =>
- val rs = (0 until runs).filter { r =>
- rand.nextDouble() < 2.0 * c(r) * k / sumCosts(r)
- }
- if (rs.nonEmpty) Some((p, rs)) else None
- }
+ pointCosts.filter { case (_, c) => rand.nextDouble() < 2.0 * c * k / sumCosts }.map(_._1)
}.collect()
- mergeNewCenters()
- chosen.foreach { case (p, rs) =>
- rs.foreach(newCenters(_) += p.toDense)
- }
+ newCenters = chosen.map(_.toDense)
+ centers ++= newCenters
step += 1
}
- mergeNewCenters()
costs.unpersist(blocking = false)
bcNewCentersList.foreach(_.destroy(false))
- // Finally, we might have a set of more than k candidate centers for each run; weigh each
+ if (centers.size <= k) {
+ return centers.toArray
--- End diff --
I prefer to avoid the `return` keyword and just put the other code under the else here. But it is a small preference.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/15342
Otherwise updated to reflect all the other review comments, thanks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by yanboliang <gi...@git.apache.org>.
Github user yanboliang commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82735096
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -499,13 +414,38 @@ object KMeans {
* @param data Training points as an `RDD` of `Vector` types.
* @param k Number of clusters to create.
* @param maxIterations Maximum number of iterations allowed.
+ * @param initializationMode The initialization algorithm. This can either be "random" or
+ * "k-means||". (default: "k-means||")
+ * @param seed Random seed for cluster initialization. Default is to generate seed based
+ * on system time.
+ */
+ @Since("2.1.0")
+ def train(data: RDD[Vector],
--- End diff --
+1 @sethah
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by yanboliang <gi...@git.apache.org>.
Github user yanboliang commented on the issue:
https://github.com/apache/spark/pull/15342
I'm more prefer to maintain the original logic that keep ```model.clusterCenters.length``` equal to ```k```. Was there some discussion as to make this change?
I checked popular Python machine learning library scikit-learn, it return the requested number of centroids:
And for R kmeans, it throw error if there is more cluster centers than distinct data points:
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMean...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r81856398
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -558,6 +475,7 @@ object KMeans {
* Trains a k-means model using specified parameters and the default values for unspecified.
*/
@Since("0.8.0")
+ @deprecated("Use train method without 'runs'", "2.1.0")
--- End diff --
There are two other `train` signatures that use `runs`, but have not been marked as deprecated.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by yanboliang <gi...@git.apache.org>.
Github user yanboliang commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82624159
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
+ totalContribs.foreach { case (j, (sum, count)) =>
+ scal(1.0 / count, sum)
+ val newCenter = new VectorWithNorm(sum)
+ if (converged && KMeans.fastSquaredDistance(newCenter, centers(j)) > epsilon * epsilon) {
+ converged = false
}
- costs(run) = costAccums(i).value
+ centers(j) = newCenter
}
- activeRuns = activeRuns.filter(active(_))
+ cost = costAccum.value
iteration += 1
}
val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
- logInfo(s"Iterations took " + "%.3f".format(iterationTimeInSeconds) + " seconds.")
+ logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")
if (iteration == maxIterations) {
logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
} else {
logInfo(s"KMeans converged in $iteration iterations.")
}
- val (minCost, bestRun) = costs.zipWithIndex.min
+ logInfo(s"The cost is $cost.")
- logInfo(s"The cost for the best run is $minCost.")
-
- new KMeansModel(centers(bestRun).map(_.vector))
+ new KMeansModel(centers.map(_.vector))
}
/**
- * Initialize `runs` sets of cluster centers at random.
+ * Initialize a set of cluster centers at random.
*/
- private def initRandom(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
- // Sample all the cluster centers in one pass to avoid repeated scans
- val sample = data.takeSample(true, runs * k, new XORShiftRandom(this.seed).nextInt()).toSeq
- Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
- new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm)
- }.toArray)
+ private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
+ data.takeSample(true, k, new XORShiftRandom(this.seed).nextInt()).map(_.toDense)
--- End diff --
Is it necessary to cast vector to dense one?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15342
**[Test build #66381 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66381/consoleFull)** for PR 15342 at commit [`ebbb852`](https://github.com/apache/spark/commit/ebbb852ad69f3a0a9c9facbd4a55ecce0eb86df0).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15342: [SPARK-11560] [MLLIB] Optimize KMeans implementat...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/15342#discussion_r82681907
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala ---
@@ -258,149 +252,106 @@ class KMeans private (
}
}
val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
- logInfo(s"Initialization with $initializationMode took " + "%.3f".format(initTimeInSeconds) +
- " seconds.")
+ logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")
- val active = Array.fill(numRuns)(true)
- val costs = Array.fill(numRuns)(0.0)
-
- var activeRuns = new ArrayBuffer[Int] ++ (0 until numRuns)
+ var converged = false
+ var cost = 0.0
var iteration = 0
val iterationStartTime = System.nanoTime()
- instr.foreach(_.logNumFeatures(centers(0)(0).vector.size))
+ instr.foreach(_.logNumFeatures(centers.head.vector.size))
- // Execute iterations of Lloyd's algorithm until all runs have converged
- while (iteration < maxIterations && !activeRuns.isEmpty) {
- type WeightedPoint = (Vector, Long)
- def mergeContribs(x: WeightedPoint, y: WeightedPoint): WeightedPoint = {
- axpy(1.0, x._1, y._1)
- (y._1, x._2 + y._2)
- }
-
- val activeCenters = activeRuns.map(r => centers(r)).toArray
- val costAccums = activeRuns.map(_ => sc.doubleAccumulator)
-
- val bcActiveCenters = sc.broadcast(activeCenters)
+ // Execute iterations of Lloyd's algorithm until converged
+ while (iteration < maxIterations && !converged) {
+ val costAccum = sc.doubleAccumulator
+ val bcCenters = sc.broadcast(centers)
// Find the sum and count of points mapping to each center
val totalContribs = data.mapPartitions { points =>
- val thisActiveCenters = bcActiveCenters.value
- val runs = thisActiveCenters.length
- val k = thisActiveCenters(0).length
- val dims = thisActiveCenters(0)(0).vector.size
+ val thisCenters = bcCenters.value
+ val dims = thisCenters.head.vector.size
- val sums = Array.fill(runs, k)(Vectors.zeros(dims))
- val counts = Array.fill(runs, k)(0L)
+ val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
+ val counts = Array.fill(thisCenters.length)(0L)
points.foreach { point =>
- (0 until runs).foreach { i =>
- val (bestCenter, cost) = KMeans.findClosest(thisActiveCenters(i), point)
- costAccums(i).add(cost)
- val sum = sums(i)(bestCenter)
- axpy(1.0, point.vector, sum)
- counts(i)(bestCenter) += 1
- }
+ val (bestCenter, cost) = KMeans.findClosest(thisCenters, point)
+ costAccum.add(cost)
+ val sum = sums(bestCenter)
+ axpy(1.0, point.vector, sum)
+ counts(bestCenter) += 1
}
- val contribs = for (i <- 0 until runs; j <- 0 until k) yield {
- ((i, j), (sums(i)(j), counts(i)(j)))
- }
- contribs.iterator
- }.reduceByKey(mergeContribs).collectAsMap()
-
- bcActiveCenters.destroy(blocking = false)
-
- // Update the cluster centers and costs for each active run
- for ((run, i) <- activeRuns.zipWithIndex) {
- var changed = false
- var j = 0
- while (j < k) {
- val (sum, count) = totalContribs((i, j))
- if (count != 0) {
- scal(1.0 / count, sum)
- val newCenter = new VectorWithNorm(sum)
- if (KMeans.fastSquaredDistance(newCenter, centers(run)(j)) > epsilon * epsilon) {
- changed = true
- }
- centers(run)(j) = newCenter
- }
- j += 1
- }
- if (!changed) {
- active(run) = false
- logInfo("Run " + run + " finished in " + (iteration + 1) + " iterations")
+ counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
+ }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
+ axpy(1.0, sum2, sum1)
+ (sum1, count1 + count2)
+ }.collectAsMap()
+
+ bcCenters.destroy(blocking = false)
+
+ // Update the cluster centers and costs
+ converged = true
+ totalContribs.foreach { case (j, (sum, count)) =>
+ scal(1.0 / count, sum)
+ val newCenter = new VectorWithNorm(sum)
+ if (converged && KMeans.fastSquaredDistance(newCenter, centers(j)) > epsilon * epsilon) {
+ converged = false
}
- costs(run) = costAccums(i).value
+ centers(j) = newCenter
}
- activeRuns = activeRuns.filter(active(_))
+ cost = costAccum.value
iteration += 1
}
val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
- logInfo(s"Iterations took " + "%.3f".format(iterationTimeInSeconds) + " seconds.")
+ logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")
if (iteration == maxIterations) {
logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
} else {
logInfo(s"KMeans converged in $iteration iterations.")
}
- val (minCost, bestRun) = costs.zipWithIndex.min
+ logInfo(s"The cost is $cost.")
- logInfo(s"The cost for the best run is $minCost.")
-
- new KMeansModel(centers(bestRun).map(_.vector))
+ new KMeansModel(centers.map(_.vector))
}
/**
- * Initialize `runs` sets of cluster centers at random.
+ * Initialize a set of cluster centers at random.
*/
- private def initRandom(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
- // Sample all the cluster centers in one pass to avoid repeated scans
- val sample = data.takeSample(true, runs * k, new XORShiftRandom(this.seed).nextInt()).toSeq
- Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
- new VectorWithNorm(Vectors.dense(v.vector.toArray), v.norm)
- }.toArray)
+ private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
+ data.takeSample(true, k, new XORShiftRandom(this.seed).nextInt()).map(_.toDense)
}
/**
- * Initialize `runs` sets of cluster centers using the k-means|| algorithm by Bahmani et al.
+ * Initialize a set of cluster centers using the k-means|| algorithm by Bahmani et al.
* (Bahmani et al., Scalable K-Means++, VLDB 2012). This is a variant of k-means++ that tries
- * to find with dissimilar cluster centers by starting with a random center and then doing
+ * to find dissimilar cluster centers by starting with a random center and then doing
* passes where more centers are chosen with probability proportional to their squared distance
* to the current cluster set. It results in a provable approximation to an optimal clustering.
*
* The original paper can be found at http://theory.stanford.edu/~sergei/papers/vldb12-kmpar.pdf.
*/
- private def initKMeansParallel(data: RDD[VectorWithNorm])
- : Array[Array[VectorWithNorm]] = {
+ private def initKMeansParallel(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
// Initialize empty centers and point costs.
- val centers = Array.tabulate(runs)(r => ArrayBuffer.empty[VectorWithNorm])
- var costs = data.map(_ => Array.fill(runs)(Double.PositiveInfinity))
+ var costs = data.map(_ => Double.PositiveInfinity)
- // Initialize each run's first center to a random point.
+ // Initialize the first center to a random point.
val seed = new XORShiftRandom(this.seed).nextInt()
- val sample = data.takeSample(true, runs, seed).toSeq
+ val sample = data.takeSample(false, 1, seed)
// Could be empty if data is empty; fail with a better message early:
- require(sample.size >= runs, s"Required $runs samples but got ${sample.size} from $data")
- val newCenters = Array.tabulate(runs)(r => ArrayBuffer(sample(r).toDense))
-
- /** Merges new centers to centers. */
- def mergeNewCenters(): Unit = {
- var r = 0
- while (r < runs) {
- centers(r) ++= newCenters(r)
- newCenters(r).clear()
- r += 1
- }
- }
+ require(sample.nonEmpty, s"No samples available from $data")
- // On each step, sample 2 * k points on average for each run with probability proportional
- // to their squared distance from that run's centers. Note that only distances between points
+ val centers = ArrayBuffer[VectorWithNorm]()
+ var newCenters = Seq(sample.head.toDense)
--- End diff --
It's a Seq but yeah no specific preference. Where do you see that optimization BTW? I just see it implemented for `TraversableOnce`, and that's what my IDE says it calls even when given an `Array`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/15342
I backed out the change for SPARK-3261; that part is actually tiny and separable now anyway. We can discuss that here too but wanted to split it from the main change for expediency.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15342: [SPARK-11560] [SPARK-3261] [MLLIB] Optimize KMeans imple...
Posted by sethah <gi...@git.apache.org>.
Github user sethah commented on the issue:
https://github.com/apache/spark/pull/15342
What are the circumstances that lead to duplicate cluster centers? Other than the obvious having less data than requested centers. The comment on the original JIRA said training 1.3M points asking for 10k clusters only returned 1k centers.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org