Skip to content

[SPARK-3261] [MLLIB] KMeans clusterer can return duplicate cluster centers #15450

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
srowen wants to merge 7 commits into from
+85 −65
Closed

[SPARK-3261] [MLLIB] KMeans clusterer can return duplicate cluster centers #15450

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
012979e
Return potentially fewer than k cluster centers in cases where k dist…
srowen Oct 12, 2016
85c9857
Deduplicate result of random selection without replacement to match b…
srowen Oct 19, 2016
ebebcb9
Add more tests
srowen Oct 20, 2016
793e4d5
Use only distinct outputs of initial k-means || selection; update sca…
srowen Oct 21, 2016
d1004d9
More test changes; fix doc problems
srowen Oct 25, 2016
79c84ad
Further refine tests; remove deprecated runs param usages
srowen Oct 25, 2016
f870fe9
Remove last refs to 'runs'
srowen Oct 29, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,8 @@ class KMeans private (
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.
*/
@Since("1.4.0")
def getK: Int = k
Expand Down Expand Up @@ -323,7 +324,10 @@ class KMeans private (
* Initialize a set of cluster centers at random.
*/
private def initRandom(data: RDD[VectorWithNorm]): Array[VectorWithNorm] = {
data.takeSample(true, k, new XORShiftRandom(this.seed).nextInt()).map(_.toDense)
// Select without replacement; may still produce duplicates if the data has < k distinct
// points, so deduplicate the centroids to match the behavior of k-means|| in the same situation
data.takeSample(false, k, new XORShiftRandom(this.seed).nextInt())
.map(_.vector).distinct.map(new VectorWithNorm(_))
}

/**
Expand Down Expand Up @@ -378,10 +382,10 @@ class KMeans private (
costs.unpersist(blocking = false)
bcNewCentersList.foreach(_.destroy(false))

if (centers.size == k) {
if (centers.size <= k) {

@sethah sethah Oct 20, 2016 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's nothing that enforces these centers to be distinct. The following test(s) fail (IMO these are much more thorough than the existing tests):

  test("unique cluster centers") {
    val rng = new scala.util.Random(42)
    val points = (0 until 10).map(i => Vectors.dense(Array.fill(3)(rng.nextDouble)))
    val data = sc.parallelize(
      points.flatMap { point =>
        Array.fill(rng.nextInt(4))(point)
      }, 2
    )
    val norms = data.map(Vectors.norm(_, 2.0))
    val zippedData = data.zip(norms).map { case (v, norm) =>
      new VectorWithNorm(v, norm)
    }
    // less centers than k
    val km = new KMeans().setK(50)
      .setMaxIterations(10)
      .setInitializationMode("k-means||")
      .setInitializationSteps(10)
      .setSeed(42)
    val initialCenters = km.initKMeansParallel(zippedData).map(_.vector)
    assert(initialCenters.length === initialCenters.distinct.length)

    val model = km.run(data)
    val finalCenters = model.clusterCenters
    assert(finalCenters.length === finalCenters.distinct.length)

    // run local k-means
    val km2 = new KMeans().setK(10)
      .setMaxIterations(10)
      .setInitializationMode("k-means||")
      .setInitializationSteps(10)
      .setSeed(42)
    val initialCenters2 = km2.initKMeansParallel(zippedData).map(_.vector)
    assert(initialCenters2.length === initialCenters2.distinct.length)

    val model2 = km.run(data)
    val finalCenters2 = model2.clusterCenters
    assert(finalCenters2.length === finalCenters2.distinct.length)
}

(BTW I had to modify initKMeansParallel to be non-private for that test)

I guess we can call distinct on the centers before this if statement?

@srowen srowen Oct 21, 2016

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's right, although it's not a goal to guarantee distinct centers, though that's a nice to have where possible. I think my goal is improving the straightforward cases, and, maintaining as much consistency as possible. I agree with adding a test like this and adding a call to .distinct. Might as well take this to a logical conclusion.

It also raises the interesting question: if you have >= k distinct points, and happen to pick < k distinct centroids, should you go back and replenish the set of centroids? I am punting on that right now but it's a legitimate point. It's quite a corner case though.

centers.toArray
} else {
// Finally, we might have a set of more or less than k candidate centers; weight each
// Finally, we might have a set of more than k candidate centers; weight each
// candidate by the number of points in the dataset mapping to it and run a local k-means++
// on the weighted centers to pick k of them
val bcCenters = data.context.broadcast(centers)
Expand Down
34 changes: 26 additions & 8 deletions mllib/src/test/scala/org/apache/spark/mllib/clustering/KMeansSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,18 +64,34 @@ class KMeansSuite extends SparkFunSuite with MLlibTestSparkContext {
assert(model.clusterCenters.head ~== center absTol 1E-5)
}

test("no distinct points") {
test("fewer distinct points than clusters") {
val data = sc.parallelize(
Array(
Vectors.dense(1.0, 2.0, 3.0),
Vectors.dense(1.0, 2.0, 3.0),
Vectors.dense(1.0, 2.0, 3.0)),
2)
val center = Vectors.dense(1.0, 2.0, 3.0)

// Make sure code runs.
var model = KMeans.train(data, k = 2, maxIterations = 1)
assert(model.clusterCenters.size === 2)
var model = KMeans.train(data, k = 2, maxIterations = 1, initializationMode = "random")
assert(model.clusterCenters.length === 1)

model = KMeans.train(data, k = 2, maxIterations = 1, initializationMode = "k-means||")
assert(model.clusterCenters.length === 1)
}


test("fewer clusters than points") {

@sethah sethah Oct 24, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd prefer to remove these two tests since we've added a more thorough test below. We can check the "random" init method in that test as well, then we can eliminate these two.

val data = sc.parallelize(
Array(
Vectors.dense(1.0, 2.0, 3.0),
Vectors.dense(1.0, 3.0, 4.0)),
2)

var model = KMeans.train(data, k = 1, maxIterations = 1, initializationMode = "random")
assert(model.clusterCenters.length === 1)

model = KMeans.train(data, k = 1, maxIterations = 1, initializationMode = "k-means||")
assert(model.clusterCenters.length === 1)
}

test("more clusters than points") {
Expand All @@ -85,9 +101,11 @@ class KMeansSuite extends SparkFunSuite with MLlibTestSparkContext {
Vectors.dense(1.0, 3.0, 4.0)),
2)

// Make sure code runs.
var model = KMeans.train(data, k = 3, maxIterations = 1)
assert(model.clusterCenters.size === 3)
var model = KMeans.train(data, k = 3, maxIterations = 1, initializationMode = "random")
assert(model.clusterCenters.length === 2)

model = KMeans.train(data, k = 3, maxIterations = 1, initializationMode = "k-means||")
assert(model.clusterCenters.length === 2)
}

test("deterministic initialization") {
Expand Down