This is an automated email from the ASF dual-hosted git repository.
srowen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 0d35f9e [SPARK-24484][MLLIB] Power Iteration Clustering is giving
incorrect clustering results when there are mutiple leading eigen values.
0d35f9e is described below
commit 0d35f9ea3a73efd858f9e1a70fd13813ffaa344d
Author: Shahid <shahidk...@gmail.com>
AuthorDate: Tue Jan 22 18:29:18 2019 -0600
[SPARK-24484][MLLIB] Power Iteration Clustering is giving incorrect
clustering results when there are mutiple leading eigen values.
## What changes were proposed in this pull request?
**Method to determine if the top eigen values has same magnitude but
opposite signs**
The vector is written as a linear combination of the eigen vectors at
iteration k.
**So, we need to check if the reileigh coefficient at the convergence is
lesser than the norm of the estimated eigen vector before normalizing**
(Please fill in changes proposed in this fix)
Added a UT
Please review http://spark.apache.org/contributing.html before opening a
pull request.
Closes #21627 from shahidki31/picConvergence.
Authored-by: Shahid <shahidk...@gmail.com>
Signed-off-by: Sean Owen <sean.o...@databricks.com>
---
.../clustering/PowerIterationClustering.scala | 21 +++++++
.../clustering/PowerIterationClusteringSuite.scala | 68 ++++++++++++++++++++++
2 files changed, 89 insertions(+)
diff --git
a/mllib/src/main/scala/org/apache/spark/mllib/clustering/PowerIterationClustering.scala
b/mllib/src/main/scala/org/apache/spark/mllib/clustering/PowerIterationClustering.scala
index 9444f29..765f272 100644
---
a/mllib/src/main/scala/org/apache/spark/mllib/clustering/PowerIterationClustering.scala
+++
b/mllib/src/main/scala/org/apache/spark/mllib/clustering/PowerIterationClustering.scala
@@ -378,6 +378,27 @@ object PowerIterationClustering extends Logging {
logInfo(s"$msgPrefix: delta = $delta.")
diffDelta = math.abs(delta - prevDelta)
logInfo(s"$msgPrefix: diff(delta) = $diffDelta.")
+
+ if (math.abs(diffDelta) < tol) {
+ /**
+ * Power Iteration fails to converge if absolute value of top 2 eigen
values are equal,
+ * but with opposite sign. The resultant vector flip-flops between two
vectors.
+ * We should give an exception, if we detect the failure of the
convergence of the
+ * power iteration
+ */
+
+ // Rayleigh quotient = x^tAx / x^tx
+ val xTAx = curG.joinVertices(v) {
+ case (_, x, y) => x * y
+ }.vertices.values.sum()
+ val xTx = curG.vertices.mapValues(x => x * x).values.sum()
+ val rayleigh = xTAx / xTx
+
+ if (math.abs(norm - math.abs(rayleigh)) > tol) {
+ logWarning(s"Power Iteration fail to converge. delta = ${delta}," +
+ s" difference delta = ${diffDelta} and norm = ${norm}")
+ }
+ }
// update v
curG = Graph(VertexRDD(v1), g.edges)
prevDelta = delta
diff --git
a/mllib/src/test/scala/org/apache/spark/ml/clustering/PowerIterationClusteringSuite.scala
b/mllib/src/test/scala/org/apache/spark/ml/clustering/PowerIterationClusteringSuite.scala
index 0ba3ffa..97269ee 100644
---
a/mllib/src/test/scala/org/apache/spark/ml/clustering/PowerIterationClusteringSuite.scala
+++
b/mllib/src/test/scala/org/apache/spark/ml/clustering/PowerIterationClusteringSuite.scala
@@ -184,6 +184,74 @@ class PowerIterationClusteringSuite extends SparkFunSuite
assert(localAssignments === localAssignments2)
}
+ test("power iteration clustering gives incorrect results due to failed to
converge") {
+ /*
+ Graph:
+ 1
+ /
+ /
+ 0 2 -- 3
+ */
+ val data1 = spark.createDataFrame(Seq(
+ (0, 1),
+ (2, 3)
+ )).toDF("src", "dst")
+
+ val assignments1 = new PowerIterationClustering()
+ .setInitMode("random")
+ .setK(2)
+ .assignClusters(data1)
+ .select("id", "cluster")
+ .as[(Long, Int)]
+ .collect()
+ val predictions1 = Array.fill(2)(mutable.Set.empty[Long])
+ assignments1.foreach {
+ case (id, cluster) => predictions1(cluster) += id
+ }
+ assert(Set(predictions1(0).size, predictions1(1).size) !== Set(2, 2))
+
+
+ /*
+ Graph:
+ 1
+ /
+ /
+ 0 - - 2 3 -- 4
+ */
+ val data2 = spark.createDataFrame(Seq(
+ (0, 1),
+ (0, 2),
+ (3, 4)
+ )).toDF("src", "dst")
+
+ var assignments2 = new PowerIterationClustering()
+ .setInitMode("random")
+ .setK(2)
+ .assignClusters(data2)
+ .select("id", "cluster")
+ .as[(Long, Int)]
+ .collect()
+ val predictions2 = Array.fill(2)(mutable.Set.empty[Long])
+ assignments2.foreach {
+ case (id, cluster) => predictions2(cluster) += id
+ }
+ assert(Set(predictions2(0).size, predictions2(1).size) !== Set(2, 3))
+
+
+ var assignments3 = new PowerIterationClustering()
+ .setInitMode("degree")
+ .setK(2)
+ .assignClusters(data2)
+ .select("id", "cluster")
+ .as[(Long, Int)]
+ .collect()
+ val predictions3 = Array.fill(2)(mutable.Set.empty[Long])
+ assignments3.foreach {
+ case (id, cluster) => predictions3(cluster) += id
+ }
+ assert(Set(predictions3(0).size, predictions3(1).size) !== Set(2, 3))
+ }
+
test("read/write") {
val t = new PowerIterationClustering()
.setK(4)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
